CMacD/AIC_PHASE1_POC
0
1#!/usr/bin/env python2# coding: utf-83"""4TextMatch — BM25 text retrieval predictor (Phase 1, Step 2a).5 6BM25 is a ranking algorithm from information retrieval — the same family of7ideas that powers search engines. Given a query (a new product's key column8text) and a corpus (all historical key text grouped by label), it scores how9well the query matches each label and returns the top hit.10 11It works well here because product attribute labels are closely tied to the12words in the product description. A product with "DARK ROAST" in its name13should score highly against the DARK ROAST training corpus.14 15What it does16------------17 1. Builds a training corpus from the historical FINAL data. All key column18 text associated with each label is concatenated, deduplicated, and19 tokenised. This gives BM25 one document per label to score against.20 21 2. Builds a query from each new flat-file product by concatenating its key22 column values and tokenising them the same way.23 24 3. Scores every query against every label document using BM25Plus and takes25 the top result as the prediction.26 27 4. BM25 raw scores have no common unit across attributes — a score of 5.228 for BRAND means something completely different to a score of 5.2 for29 FLAVOUR. So scores are min-max scaled per attribute independently to30 bring them into the 0–1 range before being passed to Ensemble.31 32Results are stored in recom_dict as 'BM25_{attrG}' and combined with the33XGBoost predictions in Ensemble.34 35Code style36----------37Functions are written to be read straight through. Steps are broken into named38variables rather than chained. If something is not immediately obvious from the39code it has a comment. Keep it that way.40"""41 42import re43from concurrent.futures import ThreadPoolExecutor44 45import numpy as np46import pandas as pd47import nltk48from nltk.corpus import stopwords49from rank_bm25 import BM25Plus50from sklearn.preprocessing import MinMaxScaler51 52from ml_package import routing53 54# Number of candidate labels BM25 returns per product.55# Score-level fusion in Ensemble sums each method's normalised scores across56# all K candidates before taking argmax, so agreement on a label that appears57# in both methods' top-K amplifies its combined score even when it isn't each58# method's single top pick.59_BM25_TOP_K = 360 61nltk.download('punkt', quiet=True)62nltk.download('punkt_tab', quiet=True)63nltk.download('stopwords', quiet=True)64 65# ── Shared stopword list ──────────────────────────────────────────────────────66# Merged from English NLTK stopwords plus domain-specific noise tokens.67# 'chocolate' and 'original' are intentionally excluded — they are68# discriminating features (e.g. chocolate vs vanilla flavour) and their69# removal was confirmed to degrade BM25 accuracy in benchmarking.70_DOMAIN_STOPWORDS = [71 'ss', 'nan', 'unknown', 'undefined', 'company', 'category', 'llc', 'inc', 'ltd',72 'to', 'oz', 'lt', 'ct', '', 'value', 'not', 'available', 'key', 'missing',73 'label', 'may', 'great', 'from', 'of', 'for', 'null', 'nav', 'card', 'mix', 'nut',74 '&', 'kit', 'sauce', 'dish', 'cup', 'bx', 'envlp', 'can', 'bag', 'btl',75 'rfg', 'cnstr', 'unf', '*', '+', 'a', 'abc', '- .', '.', '-',76 'the', '/', "'", '(', ')', ',', '..', '....',77]78_STOPWORDS = frozenset(set(stopwords.words('english')) | set(_DOMAIN_STOPWORDS))79 80# ── Auto-NUMERIC detection ────────────────────────────────────────────────────81# Mirrors the logic in Ensemble._infer_attr_type so BM25 skips attributes82# Ensemble will route through Lookup-only anyway, avoiding wasted computation83# and spurious predictions for range-value attributes.84_NUMERIC_RE = re.compile(r'^\s*[\d#][\d\s\.\-\/]*', re.IGNORECASE)85_NUMERIC_THRESHOLD = 0.6086_NULL_STRS = {'', 'nan', 'none', 'missing', 'null', 'na',87 'NaN', 'NAN', 'NONE', 'MISSING', 'NULL', 'NA',88 'null value', 'NULL VALUE'}89 90 91def _is_numeric_attr(attr_key_cols: list, data_df: pd.DataFrame) -> bool:92 """Return True if >60% of unique non-null key-column values are numeric/range."""93 for col in attr_key_cols:94 if col not in data_df.columns:95 continue96 stripped_values = data_df[col].dropna().astype(str).str.strip()97 is_null_string = stripped_values.str.lower().isin(_NULL_STRS)98 non_null_values = stripped_values[~is_null_string].unique()99 if len(non_null_values) == 0:100 continue101 numeric_match_count = sum(bool(_NUMERIC_RE.match(v)) for v in non_null_values)102 numeric_fraction = numeric_match_count / len(non_null_values)103 if numeric_fraction >= _NUMERIC_THRESHOLD:104 return True105 return False106 107 108def _build_training_corpus(history_df: pd.DataFrame, meta_df: pd.DataFrame,109 mdm_col: str) -> pd.DataFrame:110 """111 Build a tokenised BM25 training corpus from historical FINAL data.112 113 Groups all key column values by label, deduplicates tokens, applies114 stopword removal, and tokenises with NLTK. Returns a DataFrame indexed115 by label with one '_tokenize' column per key column group.116 """117 attr_key_cols = list(meta_df.loc[meta_df['Attribute Group name'] == mdm_col,118 'Attribute Name in MDM'])119 df = history_df[attr_key_cols + [mdm_col]].copy()120 121 # symmetric_difference([mdm_col]) gives all columns except the label column.122 key_cols_only = df.columns.symmetric_difference([mdm_col])123 for col in key_cols_only:124 df[col] = df[col].replace({'&': ''}, regex=True)125 df.fillna('unknown', inplace=True)126 df.replace(re.compile(r'^\s*null\s+value\s*$', re.IGNORECASE), 'unknown', inplace=True)127 128 df = df.groupby([mdm_col]).agg(lambda x: ' '.join(x))129 df = df.apply(lambda x: x.astype(str).str.lower())130 131 # Deduplicate tokens within each cell so repeated words don't inflate BM25 IDF.132 for col in df.columns:133 df[col] = df[col].apply(lambda x: ' '.join(dict.fromkeys(x.split())))134 135 combined_col = 'X_' + mdm_col136 df[combined_col] = df.astype(str).apply(' '.join, axis=1)137 # Same dedup on the combined column after concatenating all key columns together.138 df[combined_col] = df[combined_col].apply(lambda x: ' '.join(sorted(set(x.split()))))139 140 for i, col in enumerate(attr_key_cols):141 df = df.rename(columns={col: f'X{i + 1}_{mdm_col}'})142 143 # Two-step feature build per column:144 # _text → stopword-filtered string (human-readable, used for tokenisation)145 # _tokenize → NLTK word token list (the actual BM25Plus input format)146 # Only the _tokenize columns are returned; _text is an intermediate step.147 for col in list(df.columns):148 df[col + '_text'] = df[col].apply(149 lambda x: ' '.join(w for w in x.split() if w not in _STOPWORDS)150 )151 # word_tokenize splits on punctuation as well as whitespace, which is152 # the token format BM25Plus expects.153 for col in [c for c in df.columns if 'text' in c]:154 df[col + '_tokenize'] = df[col].apply(nltk.word_tokenize)155 156 return df[[c for c in df.columns if '_tokenize' in c]]157 158 159def _build_query_corpus(flat_file_df: pd.DataFrame, meta_df: pd.DataFrame,160 mdm_col: str) -> pd.DataFrame:161 """162 Build a tokenised BM25 query corpus from the flat-file (new products).163 164 Concatenates all key column values per product row, removes stopwords,165 and tokenises. Returns a DataFrame with ITEM_DIM_KEY, key columns,166 and a 'test_tokenize' column.167 """168 attr_key_cols = list(meta_df.loc[meta_df['Attribute Group name'] == mdm_col,169 'Attribute Name in MDM'])170 df = flat_file_df[attr_key_cols + ['ITEM_DIM_KEY']].copy()171 df = df.apply(lambda x: x.astype(str).str.lower())172 df.fillna('unknown', inplace=True)173 df.replace(re.compile(r'^\s*null\s+value\s*$', re.IGNORECASE), 'unknown', inplace=True)174 175 df['test'] = df[attr_key_cols].apply(lambda row: ' '.join(row.values.astype(str)), axis=1)176 df['test'] = df['test'].replace({'&': ''}, regex=True)177 df['test_text'] = df['test'].apply(lambda x: ' '.join(w for w in x.split() if w not in _STOPWORDS))178 df['test_tokenize'] = df['test_text'].apply(nltk.word_tokenize)179 df['ITEM_DIM_KEY'] = df['ITEM_DIM_KEY'].astype(str)180 181 data = pd.merge(182 flat_file_df[['ITEM_DIM_KEY'] + attr_key_cols].astype({'ITEM_DIM_KEY': str}),183 df[['ITEM_DIM_KEY', 'test_tokenize']],184 on='ITEM_DIM_KEY', how='left',185 )186 return data[[c for c in data.columns if '_tokenize' in c] + ['ITEM_DIM_KEY'] + attr_key_cols]187 188 189def _process_one_bm25_attr(mdm_col: str, meta_df: pd.DataFrame,190 history_df: pd.DataFrame,191 flat_file_df: pd.DataFrame):192 """Process a single attribute for BM25 prediction. Returns DataFrame or None."""193 attr_key_cols = list(meta_df.loc[meta_df['Attribute Group name'] == mdm_col,194 'Attribute Name in MDM'])195 196 meta_type_vals = (197 meta_df.loc[meta_df['Attribute Group name'] == mdm_col, 'Type']198 .dropna().astype(str).str.strip().str.upper()199 .pipe(lambda s: s[~s.isin(_NULL_STRS)])200 )201 if len(meta_type_vals):202 type_val = meta_type_vals.iloc[0]203 is_vocab = type_val in ('VOCAB', 'DERIVED', 'CATEGORICAL')204 type_label = 'short fixed-list' if is_vocab else 'numeric/range'205 print(f" BM25 {mdm_col}: skipped — analyst-marked as {type_label}")206 return None207 if _is_numeric_attr(attr_key_cols, history_df):208 print(f" BM25 {mdm_col}: skipped — auto-detected as numeric/range")209 return None210 211 # Routing: skip BM25 for identity/derived composites (openness ~ 1) and212 # pathological label spaces — Lookup carries them, and BM25's dense213 # [n_products x n_labels] score matrix is a memory hazard on huge label sets.214 skip, _ = routing.skip_learned_methods(history_df, attr_key_cols, mdm_col)215 if skip:216 print(f" BM25 {mdm_col}: skipped — resolved by lookup (no modelling needed)")217 return None218 219 training_corpus = _build_training_corpus(history_df, meta_df, mdm_col)220 training_corpus = training_corpus.reset_index()221 training_corpus = training_corpus.rename(columns={222 mdm_col: 'predicted',223 f'X_{mdm_col}_text_tokenize': 'corpus',224 })225 training_corpus = training_corpus[training_corpus['corpus'].map(len) > 0]226 # Guard: all training documents reduced to empty token lists after stopword227 # removal — BM25Plus([]) would silently return zero scores for every query.228 if training_corpus.empty:229 print(f" BM25 {mdm_col}: skipped — training corpus empty after tokenisation "230 f"(all historical key values reduced to stop words or punctuation; "231 f"review input data for {mdm_col})")232 return None233 234 query_corpus = _build_query_corpus(flat_file_df, meta_df, mdm_col)235 if query_corpus.empty:236 print(f" BM25 {mdm_col}: skipped — no flat-file products have non-empty key values to score")237 return None238 239 print(f" BM25 {mdm_col}: {len(query_corpus)} products")240 241 query_corpus = (242 query_corpus243 .rename(columns={'test_tokenize': 'corpus'})244 .reset_index(drop=True)245 )246 # Deduplicate tokens in each query (corpus column holds token lists at this point).247 query_corpus['corpus'] = query_corpus['corpus'].apply(lambda x: ' '.join(sorted(set(x))))248 249 try:250 tokenized_corpus = training_corpus['corpus'].tolist()251 pred_list = training_corpus['predicted'].tolist()252 bm25 = BM25Plus(tokenized_corpus)253 254 # Vectorised score matrix: rows = products, cols = training labels.255 queries = query_corpus['corpus'].str.split().tolist()256 scores_mat = np.abs(np.vstack([bm25.get_scores(q) for q in queries]))257 pred_arr = np.array(pred_list)258 259 # Return top-K candidates per product rather than top-1.260 # Ensemble fuses BM25 and XGB by summing each method's max normalised261 # score per label, so a label appearing in both methods' top-K receives262 # a combined score even when it is not each method's single best pick.263 k = min(_BM25_TOP_K, scores_mat.shape[1])264 top_k_idx = np.argsort(-scores_mat, axis=1)[:, :k]265 top_k_scores = np.take_along_axis(scores_mat, top_k_idx, axis=1)266 267 # Expand: one row per (product, candidate) pair.268 repeated = query_corpus.loc[269 query_corpus.index.repeat(k)270 ].reset_index(drop=True)271 repeated['predicted'] = pred_arr[top_k_idx.ravel()]272 repeated['max_score'] = top_k_scores.ravel()273 repeated['document'] = mdm_col274 275 return repeated.fillna('')276 except Exception as exc:277 print(f" BM25 {mdm_col}: ERROR during scoring — {exc}")278 return None279 280 281def _run_bm25(meta_df: pd.DataFrame, history_df: pd.DataFrame,282 flat_file_df: pd.DataFrame) -> pd.DataFrame:283 """284 Run BM25 prediction for all eligible attributes in parallel and return a285 combined result DataFrame with columns: corpus, ITEM_DIM_KEY, predicted,286 max_score, document, plus per-attribute key columns.287 288 Attributes flagged as NUMERIC or VOCAB in META are skipped — Ensemble289 routes these through Lookup-only or vocab matching respectively.290 Attributes are processed in parallel across 4 workers (numpy releases the291 GIL so threading is effective here).292 """293 attrs = meta_df['Attribute Group name'].unique().tolist()294 results_frames = []295 296 with ThreadPoolExecutor(max_workers=4) as executor:297 futures = [298 executor.submit(_process_one_bm25_attr, col, meta_df, history_df, flat_file_df)299 for col in attrs300 ]301 for future in futures:302 result = future.result()303 if result is not None:304 results_frames.append(result)305 306 all_attr_cols = list(meta_df['Attribute Name in MDM'].unique())307 empty_frame = pd.DataFrame(308 columns=['corpus', 'ITEM_DIM_KEY', 'score', 'predicted', 'max_score', 'document'] + all_attr_cols309 )310 return pd.concat([empty_frame] + results_frames, ignore_index=True)311 312 313def runTextMatch(meta_df: pd.DataFrame, history_df: pd.DataFrame,314 flat_file_df: pd.DataFrame, recom_dict: dict) -> dict:315 """316 Run BM25 predictions for all eligible attributes and store results in317 recom_dict as 'BM25_{attrG}'.318 319 Scores are min-max scaled per attribute independently — a BM25 score320 of 0.5 for BRAND is on a completely different absolute scale to 0.5 for321 FLAVOR, so they must not be normalised together.322 323 Parameters324 ----------325 meta_df : META sheet DataFrame.326 history_df : Historical FINAL data (training reference).327 flat_file_df : New products flat file (query set).328 recom_dict : Accumulator dict; BM25 results added as 'BM25_{attrG}'.329 330 Returns331 -------332 recom_dict (updated in-place and returned).333 """334 bm25_results = _run_bm25(meta_df, history_df, flat_file_df)335 bm25_results['ITEM_DIM_KEY'] = pd.to_numeric(bm25_results['ITEM_DIM_KEY'])336 337 # Scale per attribute independently — BM25 raw scores have no common unit338 # across attributes (BRAND scores are on a completely different scale to339 # FLAVOUR scores), so each attribute's scores are min-max scaled to [0, 1]340 # separately before being stored as prob_score.341 # Note: fit_transform inside groupby.transform is intentional here — each342 # group (attribute) needs its own min/max, not a global fit.343 scaler = MinMaxScaler()344 bm25_results['prob_score'] = (345 bm25_results.groupby('document')['max_score']346 .transform(lambda x: scaler.fit_transform(x.values.reshape(-1, 1)).ravel())347 )348 349 for mdm_col in meta_df['Attribute Group name'].unique():350 attr_key_cols = list(meta_df.loc[meta_df['Attribute Group name'] == mdm_col,351 'Attribute Name in MDM'])352 recom_dict['BM25_' + mdm_col] = (353 bm25_results[bm25_results['document'] == mdm_col][354 attr_key_cols + ['predicted', 'max_score', 'prob_score']355 ].rename(columns={'max_score': 'score', 'predicted': mdm_col})356 )357 358 return recom_dict359