model-3/scripts/cascade_check.py

"""Does reranking only the top-M candidates per entity cost any F0.5?

The test stage reranks all ~45 candidates/entity (80M pairs, ~12 h). But the answer
averages 3.46 matches and expected-F0.5 selection stops early on a precision-weighted
metric, so deep-rank candidates may never be selected. If so, reranking only the
top-M by a RERANKER-FREE ranker is free speed.

Method (paired - same entities, same models, only the reranking scope differs):
  full      : p = iso(lgbm_ce(features + ce_prob))            for every candidate
  cascade-M : rank by out-of-fold lgbm_noce; keep top-M/entity;
              those keep p; the rest are dropped (never rerankable, so never chosen)
  then identical calibration / conflict resolution / expected-F0.5 selection.

Paired on the same entities, so the comparison is far more sensitive than two
independent runs. Ship only at delta == 0.
"""
import os, sys, time, json
import numpy as np
import pandas as pd
import joblib
import lightgbm as lgb
from sklearn.isotonic import IsotonicRegression

sys.path.insert(0, "src")
import config as cfgmod
from config import CFG, EMB, PATHS, emb_dir
import blocking as B
import data as D
import decide as DC
import text_utils as T
import emb_source
from encoder import load_matrix, read_ids, Encoder
from features import compute_features
from evaluate import macro_f05

N_VAL = int(sys.argv[1]) if len(sys.argv) > 1 else 10000
GPU = sys.argv[2] if len(sys.argv) > 2 else "cuda:0"
if not str(GPU).startswith("cuda") and GPU != "cpu":
    GPU = f"cuda:{int(GPU)}"
MS = [4, 6, 8, 10, 12, 16, 20, 25]
ART = "work/results_A"
t0 = time.time()
log = lambda m: print(f"[{time.time()-t0:6.0f}s] {m}", flush=True)

bundle = joblib.load(os.path.join(ART, "model_bundle.joblib"))
method, param, miss_mass = bundle["method"], bundle["param"], bundle["miss_mass"]
feat_cols, lgbm_ce = bundle["feat_cols"], bundle["lgbm"]
amap = json.load(open(os.path.join(ART, "abbrev_map.json")))
ce_dir = bundle["ce_dir"]
log(f"bundle: {method}({param}) miss_mass={miss_mass:.4f} feats={len(feat_cols)}")

log("loading train data")
s1, s23, gt = D.load_data(PATHS["train"], with_gt=True)
s1["name_core_abbr"] = [T.apply_abbrev(x, amap) for x in s1["name_core"].values]
s23["name_core_abbr"] = [T.apply_abbrev(x, amap) for x in s23["name_core"].values]

def vals(df):
    v = {c: df[c].values.astype(object) for c in
         ["entity_id", "name_core_abbr", "name_norm", "address_abbrev_norm",
          "legal_suffix", "country_norm"]}
    nums = df["address_numbers"].values.astype(object)
    v["nums"] = np.where(nums == "", df["business_address"].values.astype(object), nums)
    return v
s1v, s23v = vals(s1), vals(s23)

pool_pos = {e: i for i, e in enumerate(s23["entity_id"].values)}
s1 = D.augment_s1(s1, gt)            # gt is a DataFrame; this is how the pipeline does it
lists = s1["matched_entity_ids"].map(D.split_ids)
truth_rows = {}
for r, ms in enumerate(lists):
    if ms:
        truth_rows[r] = set(pool_pos[m] for m in ms if m in pool_pos)
_npairs = sum(len(v) for v in truth_rows.values())
log(f"ground truth: {len(truth_rows):,} entities with matches, {_npairs:,} true pairs")
assert _npairs > 5_000_000, f"ground truth looks wrong ({_npairs} pairs) - expected ~7.6M"

rng = np.random.RandomState(CFG["seed"])
ce_rows, st_rows, va_rows = D.make_split(s1, CFG, rng)
va_rows = np.sort(rng.choice(va_rows, min(N_VAL, len(va_rows)), replace=False))
log(f"using {len(va_rows):,} val entities (same split/seed as the real run)")

# ------------------------------------------------------------------ candidates
c_emb, cmeta = load_matrix(emb_dir("train_pool"))
log(f"pool emb {c_emb.shape}")
enc = Encoder(CFG["emb_model"], GPU, EMB["max_len"], EMB["batch"])
q_emb = enc.encode([B.record_text(s1["business_name"].values[i],
                                  s1["business_address"].values[i],
                                  s1["country"].values[i]) for i in va_rows],
                   CFG["emb_query_prefix"])
del enc
emb_s, emb_i = B.dense_topk(np.asarray(q_emb), c_emb, CFG, GPU)

emb2_s = emb2_i = None
if CFG.get("emb2_enabled"):
    c2, q2 = emb_source.load_emb2("train", len(s1), len(s23))
    cfg2 = dict(CFG); cfg2["emb_k"] = CFG["emb2_k"]
    emb2_s, emb2_i = B.dense_topk(np.asarray(q2[va_rows]), c2, cfg2, GPU)
    del c2, q2
log("dense channels done")

lidx = B.LexicalIndex([B.lexical_tokens(n, a) for n, a in
                       zip(s23v["name_core_abbr"], s23v["address_abbrev_norm"])], CFG)
lex = B.lexical_topk(lidx, [B.lexical_tokens(s1v["name_core_abbr"][i],
                                             s1v["address_abbrev_norm"][i])
                            for i in va_rows], CFG["lex_n_jobs"])
del lidx
cand = B.build_candidates(va_rows, emb_s, emb_i, lex, s1v["country_norm"],
                          s23v["country_norm"], CFG["country_filter"],
                          emb2_s=emb2_s, emb2_i=emb2_i)
qpos = {r: i for i, r in enumerate(va_rows.tolist())}
cand = B.fill_missing_emb(cand, lambda rows: q_emb[[qpos[r] for r in rows]], c_emb)
cand["y"] = np.fromiter((c in truth_rows.get(q, ()) for q, c in
                         zip(cand["q"].values, cand["c"].values)), np.int8, len(cand))
log(f"{len(cand):,} candidates ({len(cand)/len(va_rows):.1f}/entity)")

feats = compute_features(cand, s1v, s23v, CFG["feat_n_jobs"])
cand = pd.concat([cand, feats], axis=1); del feats
log("features done")

# ------------------------------------------------------------------ reranker
from cross_encoder import pair_text, predict_cross_encoder
ta = [pair_text(s1["business_name"].values[r], s1["business_address"].values[r],
                s1["country"].values[r]) for r in cand["q"].values]
tb = [pair_text(s23["business_name"].values[c], s23["business_address"].values[c],
                s23["country"].values[c]) for c in cand["c"].values]
cand["ce_prob"] = predict_cross_encoder(ta, tb, ce_dir, CFG)
del ta, tb
log("reranker scoring done")

# ------------------------------------------------------ CE-free pre-ranker (OOF)
noce_cols = [c for c in feat_cols if c != "ce_prob"]
oof = np.zeros(len(cand))
groups = cand["q"].values
uq = np.unique(groups)
fold_of = {q: i % 5 for i, q in enumerate(rng.permutation(uq))}
fold = np.array([fold_of[q] for q in groups])
for f in range(5):
    tr, te = fold != f, fold == f
    m = lgb.train({**CFG["lgbm_params"], "verbose": -1},
                  lgb.Dataset(cand.loc[tr, noce_cols], cand.loc[tr, "y"]),
                  num_boost_round=300)
    oof[te] = m.predict(cand.loc[te, noce_cols])
cand["noce"] = oof
log("out-of-fold CE-free pre-ranker built")

# ------------------------------------------------------------------ compare
cand["p"] = lgbm_ce.predict(cand[feat_cols])
cand["p"] = bundle["calibrator"].predict(cand["p"].values)   # the real calibrator

truths = {q: truth_rows.get(q, set()) for q in va_rows}
ids = list(va_rows)

full_preds = DC.predict_sets(cand, "q", "c", "p", method, param, miss_mass, True)
m_full, per_full = macro_f05(full_preds, truths, ids)
log(f"FULL      F0.5 {m_full['f05']:.6f}  P {m_full['precision']:.4f}  R {m_full['recall']:.4f}")
if not (0.93 <= m_full["f05"] <= 0.995):
    sys.exit(f"SANITY FAIL: full-pipeline F0.5 {m_full['f05']:.4f} is nowhere near the "
             f"0.9812 the real run reported - the harness is wrong, not the cascade")

cand["rk"] = cand.groupby("q")["noce"].rank(ascending=False, method="first")
print()
print(f"{'M':>4} {'F0.5':>10} {'delta':>11} {'P':>8} {'R':>8} {'rerank%':>9} {'selected outside top-M':>24}")
print(f"{'full':>4} {m_full['f05']:>10.6f} {0.0:>11.6f} {m_full['precision']:>8.4f} "
      f"{m_full['recall']:>8.4f} {100.0:>8.1f}% {0:>24}")
for M in MS:
    sub = cand[cand["rk"] <= M]
    preds = DC.predict_sets(sub, "q", "c", "p", method, param, miss_mass, True)
    mm, per = macro_f05(preds, truths, ids)
    lost = sum(1 for q, cs in full_preds.items() for c in cs
               if c not in set(preds.get(q, [])))
    print(f"{M:>4} {mm['f05']:>10.6f} {mm['f05']-m_full['f05']:>+11.6f} "
          f"{mm['precision']:>8.4f} {mm['recall']:>8.4f} "
          f"{100*len(sub)/len(cand):>8.1f}% {lost:>24}")
print()
print(f"paired on {len(ids):,} identical entities; ship only at delta == 0")