File size: 2,782 Bytes
6861f5b | 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 | """Shared loader + feature-matrix cache (step C). Matrices cached as parquet keyed by hash(table names, feature list, sample)."""
import os, sys, hashlib, json, gc, numpy as np, polars as pl
sys.path.insert(0,'/workspace/huawei/v2work'); sys.path.insert(1,'/workspace/huawei/v3')
from feats import DROP
from features import feature_cols, GROUPS
D='/workspace/huawei/v2work'; P2=f'{D}/feat'; P3=f'{D}/feat_v3'; CACHE=f'{D}/cache_mx'; os.makedirs(CACHE,exist_ok=True)
V2_OF={'f1':'h1','f2':'h2','f3':'h3','f4':'h4','f5':'h5','f6':'val26','val25':'val25','val26':'val26','eval':'eval'}
V2_OF.update({f'{p}{k}':f'{p}{k}' for p in 'eEd' for k in range(1,7)})
TIME=['tm_stale','tm_hour','tm_tb','tm_dt','v2_tb','v2_dt','v2_stale','v2_hour']
KEEP=['rowid','eid','u','ts','clicked','us_cold','us_nohist','tm_stale']
def load_tab(n, extra=(), variant='notime'):
a=pl.read_parquet(f'{P3}/{n}.parquet'); b=pl.read_parquet(f'{P2}/{V2_OF[n]}.parquet')
f2=[c for c in b.columns if c not in DROP]
a=a.join(b.select(['rowid']+f2).rename({c:'v2_'+c for c in f2}),on='rowid',how='left')
for sub in (os.environ.get('XDIR','feat_x'),'feat_x2')+tuple(extra):
a=a.join(pl.read_parquet(f'{D}/{sub}/{n}.parquet'),on='rowid',how='left')
if '+' in variant:
import mods; a=mods.apply(a,n,variant)
return a
def featlist(df, variant='notime'):
f=feature_cols(df,GROUPS)+[c for c in df.columns if c.startswith(('v2_','x_','r_','s_','ev2_','rc_','sh_','hs_','lr'))]
if 'notime' in variant: f=[c for c in f if c not in TIME]
if variant.split('+')[0]=='stale': f=[c for c in f if c not in TIME or c=='tm_stale']
return f
def usample(df, frac):
if frac>=1: return df
return df.filter((pl.col('u').hash(seed=7) % 10000) < int(frac*10000))
def matrix(n, frac=1.0, variant='notime', extra=()):
"""returns (X float32, meta DataFrame, feats); cached."""
key=hashlib.md5(json.dumps([n,frac,variant,list(extra),os.path.getmtime(f'{P3}/{n}.parquet')]+([os.environ['XDIR']] if os.environ.get('XDIR','feat_x')!='feat_x' else [])).encode()).hexdigest()[:12]
fp=f'{CACHE}/{n}_{key}.parquet'
if os.path.exists(fp):
t=pl.read_parquet(fp); feats=[c for c in t.columns if c not in KEEP or c in ('us_cold','us_nohist')]
fl=json.load(open(fp+'.json'));
else:
t=usample(load_tab(n,extra,variant),frac); fl=featlist(t,variant)
t=t.select(list(dict.fromkeys([c for c in KEEP if c in t.columns]+fl)))
t=t.with_columns([pl.col(c).cast(pl.Float32) for c in fl if t[c].dtype!=pl.Float32 and c not in KEEP])
t.write_parquet(fp); json.dump(fl,open(fp+'.json','w'))
X=t.select(fl).to_numpy().astype(np.float32); meta=t.select([c for c in KEEP if c in t.columns]); del t; gc.collect()
return X, meta, fl
|