| """Re-ranker v2: candidate-EC features from the merged hit list (train + UniProt references). |
| |
| The same function builds the training rows (official-label proxy, see experiments/external) and the |
| inference rows, so features cannot drift. Candidates are the top 10 ECs of the weighted vote. |
| """ |
| import math |
|
|
| FEATURES = ['v12w', 'v12', 'v8', 'v4', 'v1', 'rank', 's_tr', 's_sp', 's_un', 's_u2', 'best_bits_rel', 'best_pid', |
| 'best_rank', 'best_src', 'qcov', 'scov', 'n10', 'n50', 'l3', 'l2', 'lfreq', 'top_pid', 'top_bits', |
| 'top_src', 'n_hits', 'n_cand', 'margin', 'pid_gap', 'nec_best'] |
| SRC = {'sp_': 1, 'cp_': 1, 'br_': 1, 'dk_': 1, 'bo_': 1, 'pd_': 1, 'un_': 2, 'es_': 2, 'e2_': 2, 'u2_': 3} |
|
|
|
|
| def source(s): |
| """0 = training protein (ids tr_<digits>), 1 = curated external (Swiss-Prot, BRENDA, CatPred, DLKcat), |
| 2 = unreviewed with protein-level evidence or ESP, 3 = TrEMBL transcript-level.""" |
| return 0 if (s.startswith('tr_') and s[3:].isdigit()) else SRC.get(s[:3], 1) |
|
|
|
|
| def candidate_features(hits, labels, freq, weights, k=50, power=12): |
| """hits: (sseqid, pid, qlen, slen, evalue, bits, qstart, qend, sstart, send) tuples, any order. |
| Returns (candidate ECs ordered by weighted vote, feature rows).""" |
| h = sorted(hits, key=lambda x: -x[5])[:k] |
| if not h: |
| return [], [] |
| top = h[0][5] |
| st, l3, l2 = {}, {}, {} |
| tot = {'w': 0.0, '12': 0.0, '8': 0.0, '4': 0.0, '1': 0.0} |
| for r, (s, pid, ql, sl, ev, b, qs, qe, ss, se) in enumerate(h): |
| ecs = labels[s] |
| n = len(ecs) |
| rb = b / top |
| src = source(s) |
| w = rb ** power * (1.0 if src == 0 else weights.get(s[:3], 1.0)) |
| tot['w'] += w |
| tot['12'] += rb ** 12 |
| tot['8'] += rb ** 8 |
| tot['4'] += rb ** 4 |
| tot['1'] += rb |
| for e in ecs: |
| d = st.get(e) |
| if d is None: |
| d = st[e] = dict(w=0.0, p12=0.0, p8=0.0, p4=0.0, p1=0.0, src=[0.0, 0.0, 0.0, 0.0], bb=b, bp=pid, br=r, |
| bs=src, qc=(qe - qs + 1) / ql, sc=(se - ss + 1) / sl, n10=0, n50=0, nec=n) |
| d['w'] += w / n |
| d['p12'] += rb ** 12 / n |
| d['p8'] += rb ** 8 / n |
| d['p4'] += rb ** 4 / n |
| d['p1'] += rb / n |
| d['src'][src] += w / n |
| d['n10'] += r < 10 |
| d['n50'] += 1 |
| p3 = e.rsplit('.', 1)[0] |
| p2 = '.'.join(e.split('.')[:2]) |
| l3[p3] = l3.get(p3, 0.0) + w / n |
| l2[p2] = l2.get(p2, 0.0) + w / n |
| cands = sorted(st, key=lambda e: (-st[e]['w'], -freq.get(e, 0), e))[:10] |
| ws = [st[e]['w'] for e in cands] |
| margin = (ws[0] - (ws[1] if len(ws) > 1 else 0.0)) / tot['w'] |
| rows = [] |
| for i, e in enumerate(cands): |
| d = st[e] |
| rows.append([d['w'] / tot['w'], d['p12'] / tot['12'], d['p8'] / tot['8'], d['p4'] / tot['4'], d['p1'] / tot['1'], i, |
| *[x / tot['w'] for x in d['src']], d['bb'] / top, d['bp'], d['br'], d['bs'], d['qc'], d['sc'], |
| d['n10'], d['n50'], l3[e.rsplit('.', 1)[0]] / tot['w'], l2['.'.join(e.split('.')[:2])] / tot['w'], |
| math.log1p(freq.get(e, 0)), h[0][1], top, source(h[0][0]), len(h), len(st), margin, |
| h[0][1] - d['bp'], d['nec']]) |
| return cands, rows |
|
|
