| """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') |
| |
| 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+1<len(roots) else len(nodes) |
| active=np.arange(len(x));idx=np.full(len(x),root,dtype=np.int64);steps=0 |
| while len(active): |
| if steps>end-root:raise ContractError('Tree cycle detected') |
| if (idx<root).any() or (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) |
|
|
