File size: 8,234 Bytes
6d5d7e5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | """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+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)
|