"""Data-only execution of parameters exported from the user's fitted models. Normal runtime never deserializes pickle/joblib or imports arbitrary model code. The one-time isolated exporter and its source hashes are delivered separately. """ from __future__ import annotations from pathlib import Path import hashlib,json,math,re,zipfile from typing import Any, Mapping import numpy as np class ContractError(ValueError): pass def file_sha256(path: Path) -> str: h=hashlib.sha256() with Path(path).open('rb') as f: for chunk in iter(lambda:f.read(1024*1024),b''):h.update(chunk) return h.hexdigest() class ParameterPack: def __init__(self, root: Path, expected_manifest_sha256: str): root=Path(root).resolve();manifest=root/'manifest.json' if not re.fullmatch(r'[0-9a-f]{64}',expected_manifest_sha256):raise ContractError('A pinned manifest hash is required') if manifest.stat().st_size>2*1024*1024 or file_sha256(manifest)!=expected_manifest_sha256:raise ContractError('Manifest integrity mismatch') self.manifest=json.loads(manifest.read_text(encoding='utf-8')) if self.manifest.get('schema')!='humanization.fitted-parameter-pack/1':raise ContractError('Unsupported pack schema') name=self.manifest.get('arrays_file') if name!='parameters.npz':raise ContractError('Unsafe parameter filename') path=root/name if path.is_symlink() or path.stat().st_size>64*1024**2 or file_sha256(path)!=self.manifest['arrays_sha256']:raise ContractError('Parameter integrity mismatch') with zipfile.ZipFile(path) as z: entries=z.infolist() if len(entries)!=self.manifest['array_count'] or len({i.filename for i in entries})!=len(entries):raise ContractError('Array count/duplicate mismatch') if sum(i.file_size for i in entries)>128*1024**2 or any(not re.fullmatch(r'a\d{5}\.npy',i.filename) or i.file_size>32*1024**2 for i in entries):raise ContractError('Parameter archive bounds exceeded') with np.load(path,allow_pickle=False) as z: self.arrays={k:z[k] for k in z.files} if any(a.dtype.hasobject for a in self.arrays.values()):raise ContractError('Executable/object arrays are forbidden') for a in self.arrays.values():a.setflags(write=False) self.data=self._decode(self.manifest['data']) self.id=expected_manifest_sha256 def _decode(self,x: Any) -> Any: if isinstance(x,dict) and 'array' in x: a=self.arrays[x['array']] if list(a.shape)!=x['shape'] or str(a.dtype)!=x['dtype']:raise ContractError('Array shape/dtype mismatch') return a if isinstance(x,dict):return {k:self._decode(v) for k,v in x.items()} if isinstance(x,list):return [self._decode(v) for v in x] return x def feature_row(features: Mapping[str, Any], names: list[str], dtype=np.float64) -> np.ndarray: if not isinstance(features,dict):raise ContractError('features must be an object') missing=[k for k in names if k not in features];extra=sorted(set(features)-set(names)) if missing or extra:raise ContractError(f'Feature schema mismatch: missing={missing}; unknown={extra}') values=[] for k in names: v=features[k] if isinstance(v,(bool,str)) or not isinstance(v,(int,float,np.integer,np.floating)) or not math.isfinite(float(v)):raise ContractError(f'{k}: finite numeric value required; no implicit zero/median imputation') values.append(v) with np.errstate(over='ignore',invalid='ignore'):row=np.asarray([values],dtype=dtype) if not np.isfinite(row).all():raise ContractError('Input exceeds numeric representation') return row def matrix(x: Any,n: int)->np.ndarray: x=np.asarray(x) if x.ndim!=2 or x.shape[1]!=n or not 1<=len(x)<=4096 or x.dtype.kind not in 'fiu' or not np.isfinite(x).all():raise ContractError(f'Finite numeric matrix with {n} columns and 1..4096 rows required') return x def transform(model: dict,x: np.ndarray)->np.ndarray: x=matrix(x,model['n_features']).copy() if x.dtype.kind!='f':x=x.astype(np.float64) if model['kind']!='standard_scaler':raise ContractError('Not a scaler') # Preserve the original sklearn in-place float32 rounding semantics. if model['with_mean']:x-=np.asarray(model['mean'],dtype=x.dtype) if model['with_std']:x/=np.asarray(model['scale'],dtype=x.dtype) if not np.isfinite(x).all():raise ContractError('Non-finite scaling result') return x def _hgb_raw(model: dict,x: np.ndarray)->np.ndarray: x=matrix(x,model['n_features']);nodes=model['nodes'];roots=model['roots'];channels=model['output_channels'] out=np.repeat(np.asarray(model['baseline'],dtype=np.float64),len(x),axis=0) for i,root in enumerate(roots): root=int(root);end=int(roots[i+1]) if i+1end-root:raise ContractError('Tree cycle detected') if (idx=end).any():raise ContractError('Invalid tree child') current=nodes[idx];leaf=current['leaf'];done=active[leaf] out[done,int(channels[i])]+=current['value'][leaf] active=active[~leaf];current=current[~leaf] if len(active): f=current['feature'] if (f<0).any() or (f>=x.shape[1]).any():raise ContractError('Invalid tree feature') values=x[active,f] left=np.where(np.isnan(values),current['missing_left'],values<=current['threshold']) idx=root+np.where(left,current['left'],current['right']) steps+=1 return out def predict_proba(model: dict,x: np.ndarray)->np.ndarray: kind=model['kind'];x=matrix(x,model['n_features']) if kind=='hgb_classifier': raw=_hgb_raw(model,x) if len(model['classes'])==2 and raw.shape[1]==1: p=np.exp(-np.logaddexp(0,-raw[:,0]));return np.column_stack([1-p,p]) raw=raw-np.max(raw,axis=1,keepdims=True);p=np.exp(raw);return p/p.sum(axis=1,keepdims=True) if kind=='logistic' and len(model['classes'])==2: raw=(x@model['coef'].T+model['intercept'])[:,0] p=np.exp(-np.logaddexp(0,-raw));return np.column_stack([1-p,p]) raise ContractError('Unsupported probability model') def predict(model: dict,x: np.ndarray)->np.ndarray: kind=model['kind'];x=matrix(x,model['n_features']) if kind=='ridge':out=x@model['coef'].T+model['intercept'] elif kind=='hgb_regressor':out=_hgb_raw(model,x)[:,0] elif kind in ('hgb_classifier','logistic'):return np.asarray(model['classes'])[np.argmax(predict_proba(model,x),axis=1)] else:raise ContractError('Unsupported estimator kind') if not np.isfinite(out).all():raise ContractError('Non-finite prediction') return out def isolation_score(model:dict,x:np.ndarray)->np.ndarray: x=matrix(x,model['n_features']).astype(np.float32);depths=np.zeros(len(x),dtype=np.float64) for t in model['trees']: xx=x[:,t['features_subset']];active=np.arange(len(x));idx=np.zeros(len(x),dtype=np.int64);steps=0 while len(active): if steps>len(t['left']):raise ContractError('Isolation tree cycle') if (idx<0).any() or (idx>=len(t['left'])).any():raise ContractError('Isolation tree child') leaf=t['left'][idx]==-1 depths[active[leaf]]+=t['depth'][idx[leaf]]+t['average_path_length'][idx[leaf]]-1.0 current=idx[~leaf];active=active[~leaf] if len(active): f=t['feature'][current] if (f<0).any() or (f>=xx.shape[1]).any():raise ContractError('Isolation feature') idx=np.where(xx[active,f]<=t['threshold'][current],t['left'][current],t['right'][current]) steps+=1 n=model['max_samples'] average=0.0 if n<=1 else 1.0 if n==2 else 2.0*(np.log(n-1)+np.euler_gamma)-2.0*(n-1)/n denominator=len(model['trees'])*average with np.errstate(divide='ignore',invalid='ignore'): exponent=np.divide(depths,denominator,out=np.ones_like(depths),where=denominator!=0) return -np.power(2.0,-exponent)