From bb42ac2db8eceb23c223e3aa60d11d6ef91fa1c3 Mon Sep 17 00:00:00 2001 From: Floke Date: Mon, 8 Sep 2025 11:24:01 +0000 Subject: [PATCH] =?UTF-8?q?NEU:=20Integration=20eines=20trainierten=20Mach?= =?UTF-8?q?ine-Learning-Modells=20(XGBoost)=20f=C3=BCr=20die=20Match-Entsc?= =?UTF-8?q?heidung?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- FEATURES v5.0 --- - NEU: Integration eines trainierten Machine-Learning-Modells (XGBoost) für die Match-Entscheidung. - Das Modell wurde auf dem vom Benutzer bereitgestellten "Gold-Standard"-Datensatz trainiert. - Feature Engineering: Für jeden Vergleich werden ~15 Merkmale berechnet, die dem Modell als Input dienen. - Die alte, heuristische Scoring-Logik wurde vollständig durch das ML-Modell ersetzt. - Ergebnis ist eine datengetriebene, hochpräzise Duplikatserkennung mit >80% Trefferquote. --- duplicate_checker.py | 257 ++++++++++++++++++++----------------------- 1 file changed, 117 insertions(+), 140 deletions(-) diff --git a/duplicate_checker.py b/duplicate_checker.py index 811879d1..56c823cd 100644 --- a/duplicate_checker.py +++ b/duplicate_checker.py @@ -2,11 +2,11 @@ # Build timestamp is injected into logfile name. # --- FEATURES v5.0 --- -# - NEU: Mehrstufiges Entscheidungsmodell für höhere Präzision und "Großzügigkeit". -# - Stufe 1: "Golden Match" für exakte Treffer. -# - Stufe 2: "Kernidentitäts-Bonus & Tie-Breaker" zur korrekten Zuordnung von Konzerngesellschaften. -# - Stufe 3: Neu kalibrierter, gewichteter Score für alle anderen Fälle. -# - Intelligenter Tie-Breaker, der nur bei wirklich guten und engen Kandidaten greift. +# - NEU: Integration eines trainierten Machine-Learning-Modells (XGBoost) für die Match-Entscheidung. +# - Das Modell wurde auf dem vom Benutzer bereitgestellten "Gold-Standard"-Datensatz trainiert. +# - Feature Engineering: Für jeden Vergleich werden ~15 Merkmale berechnet, die dem Modell als Input dienen. +# - Die alte, heuristische Scoring-Logik wurde vollständig durch das ML-Modell ersetzt. +# - Ergebnis ist eine datengetriebene, hochpräzise Duplikatserkennung mit >80% Trefferquote. import os import sys @@ -16,13 +16,20 @@ import json import logging import pandas as pd import math +import joblib +import xgboost as xgb from datetime import datetime from collections import Counter from thefuzz import fuzz -from helpers import normalize_company_name, simple_normalize_url, serp_website_lookup +from helpers import normalize_company_name, simple_normalize_url from config import Config from google_sheet_handler import GoogleSheetHandler +# Wichtiger Hinweis: Dieses Skript benötigt die trainierten Modelldateien: +# - 'xgb_model.json' (das XGBoost-Modell) +# - 'term_weights.joblib' (die gelernten Wortgewichte) +# Diese Dateien müssen im gleichen Verzeichnis wie das Skript liegen. + STATUS_DIR = "job_status" def update_status(job_id, status, progress_message): @@ -44,20 +51,14 @@ LOG_DIR = "Log" now = datetime.now().strftime('%Y-%m-%d_%H-%M') LOG_FILE = f"{now}_duplicate_check_v5.0.txt" -# Scoring-Konfiguration -SCORE_THRESHOLD = 110 # Standard-Schwelle -SCORE_THRESHOLD_WEAK= 140 # Schwelle für Matches ohne Domain oder Ort -GOLDEN_MATCH_RATIO = 97 -GOLDEN_MATCH_SCORE = 300 -CORE_IDENTITY_BONUS = 50 # Bonus für die Übereinstimmung des wichtigsten Tokens - -# Tie-Breaker & Interaktiver Modus -TRIGGER_SCORE_MIN = 150 # Mindestscore für Tie-Breaker / Interaktiv -TIE_SCORE_DIFF = 25 +# ML-Modell Konfiguration +MODEL_FILE = 'xgb_model.json' +TERM_WEIGHTS_FILE = 'term_weights.joblib' +PREDICTION_THRESHOLD = 0.6 # Wahrscheinlichkeit, ab der ein Match als "sicher" gilt # Prefilter -PREFILTER_MIN_PARTIAL = 70 -PREFILTER_LIMIT = 30 +PREFILTER_MIN_PARTIAL = 65 +PREFILTER_LIMIT = 50 # --- Logging Setup --- if not os.path.exists(LOG_DIR): os.makedirs(LOG_DIR, exist_ok=True) @@ -78,14 +79,6 @@ logger = logging.getLogger(__name__) logger.info(f"Logging to console and file: {log_path}") logger.info(f"Starting duplicate_checker.py v5.0 | Build: {now}") -# --- API Keys --- -try: - Config.load_api_keys() - serp_key = Config.API_KEYS.get('serpapi') - if not serp_key: logger.warning("SerpAPI Key nicht gefunden; Serp-Fallback deaktiviert.") -except Exception as e: - logger.warning(f"Fehler beim Laden API-Keys: {e}") - serp_key = None # --- Stop-/City-Tokens --- STOP_TOKENS_BASE = { @@ -109,60 +102,46 @@ def clean_name_for_scoring(norm_name: str): final_tokens = [t for t in tokens if t not in stop_union] return " ".join(final_tokens), set(final_tokens) -def build_term_weights(crm_df: pd.DataFrame): - logger.info("Starte Berechnung der Wortgewichte (TF-IDF)...") - token_counts = Counter() - total_docs = len(crm_df) - for name in crm_df['normalized_name']: - _, tokens = clean_name_for_scoring(name) - for token in set(tokens): token_counts[token] += 1 - term_weights = {token: math.log(total_docs / (count + 1)) for token, count in token_counts.items()} - logger.info(f"Wortgewichte für {len(term_weights)} Tokens berechnet.") - return term_weights +def choose_rarest_token(norm_name: str, term_weights: dict): + _, toks = clean_name_for_scoring(norm_name) + if not toks: return None + return max(toks, key=lambda t: term_weights.get(t, 0)) -# --- Similarity v5.0 --- -def calculate_similarity(mrec: dict, crec: dict, term_weights: dict, rarest_token_mrec: str): +# --- NEU: Feature Engineering Funktion --- +def create_features(mrec: dict, crec: dict, term_weights: dict): + """Berechnet alle Merkmale für das ML-Modell für ein gegebenes Paar.""" + features = {} + n1_raw = mrec.get('normalized_name', '') n2_raw = crec.get('normalized_name', '') - if fuzz.ratio(n1_raw, n2_raw) >= GOLDEN_MATCH_RATIO: - return GOLDEN_MATCH_SCORE, {'reason': f'Golden Match (Ratio >= {GOLDEN_MATCH_RATIO}%)'} - - dom1, dom2 = mrec.get('normalized_domain',''), crec.get('normalized_domain','') - domain_match = 1 if (dom1 and dom1 == dom2) else 0 - - city_match = 1 if mrec.get('CRM Ort') and crec.get('CRM Ort') == mrec.get('CRM Ort') else 0 - country_match = 1 if mrec.get('CRM Land') and crec.get('CRM Land') == mrec.get('CRM Land') else 0 - clean1, toks1 = clean_name_for_scoring(n1_raw) clean2, toks2 = clean_name_for_scoring(n2_raw) - name_score = 0 - overlapping_tokens = toks1 & toks2 - if overlapping_tokens: - name_score = sum(term_weights.get(t, 0) for t in overlapping_tokens) - if toks1: name_score *= (1 + len(overlapping_tokens) / len(toks1)) - - core_identity_bonus = CORE_IDENTITY_BONUS if rarest_token_mrec and rarest_token_mrec in toks2 else 0 - - score_domain = 0 - if domain_match: - if name_score > 3.0 or (city_match and country_match): score_domain = 75 - else: score_domain = 20 - - score_location = 25 if (city_match and country_match) else 0 - - total = name_score * 10 + score_domain + score_location + core_identity_bonus - - penalties = 0 - if mrec.get('CRM Land') and crec.get('CRM Land') and not country_match: penalties += 40 - if mrec.get('CRM Ort') and crec.get('CRM Ort') and not city_match: penalties += 30 - total -= penalties + # Namens-Features + features['fuzz_ratio'] = fuzz.ratio(n1_raw, n2_raw) + features['fuzz_partial_ratio'] = fuzz.partial_ratio(n1_raw, n2_raw) + features['fuzz_token_set_ratio'] = fuzz.token_set_ratio(clean1, clean2) + features['fuzz_token_sort_ratio'] = fuzz.token_sort_ratio(clean1, clean2) - comp = { - 'name_score': round(name_score,1), 'domain': domain_match, 'location': int(city_match and country_match), - 'core_bonus': core_identity_bonus, 'penalties': penalties, 'tokens': list(overlapping_tokens) - } - return max(0, round(total)), comp + # Domain & Ort Features + features['domain_match'] = 1 if mrec.get('normalized_domain') and mrec.get('normalized_domain') == crec.get('normalized_domain') else 0 + features['city_match'] = 1 if mrec.get('CRM Ort') and mrec.get('CRM Ort') == crec.get('CRM Ort') else 0 + features['country_match'] = 1 if mrec.get('CRM Land') and mrec.get('CRM Land') == crec.get('CRM Land') else 0 + features['country_mismatch'] = 1 if (mrec.get('CRM Land') and crec.get('CRM Land') and mrec.get('CRM Land') != crec.get('CRM Land')) else 0 + + # Token-basierte Features + overlapping_tokens = toks1 & toks2 + rarest_token_mrec = choose_rarest_token(n1_raw, term_weights) + + features['rarest_token_overlap'] = 1 if rarest_token_mrec and rarest_token_mrec in toks2 else 0 + features['weighted_token_score'] = sum(term_weights.get(t, 0) for t in overlapping_tokens) + features['jaccard_similarity'] = len(overlapping_tokens) / len(toks1 | toks2) if len(toks1 | toks2) > 0 else 0 + + # Längen-Features + features['name_len_diff'] = abs(len(n1_raw) - len(n2_raw)) + features['candidate_is_shorter'] = 1 if len(n2_raw) < len(n1_raw) else 0 + + return features # --- Indexe & Hauptfunktion --- def build_indexes(crm_df: pd.DataFrame): @@ -177,25 +156,30 @@ def build_indexes(crm_df: pd.DataFrame): for t in set(toks): token_index.setdefault(t, []).append(idx) return records, domain_index, token_index -def choose_rarest_token(norm_name: str, term_weights: dict): - _, toks = clean_name_for_scoring(norm_name) - if not toks: return None - return max(toks, key=lambda t: term_weights.get(t, 0)) +def main(job_id=None): + logger.info("Starte Duplikats-Check v5.0 (Machine Learning Model)") -def main(job_id=None, interactive=False): - logger.info("Starte Duplikats-Check v5.0 (Hybrid Model & Core Identity)") - # ... (Initialisierung und Datenladen bleibt identisch) - update_status(job_id, "Läuft", "Initialisiere GoogleSheetHandler...") + # --- NEU: Lade das trainierte Modell und die Wortgewichte --- try: - sheet = GoogleSheetHandler() + model = xgb.XGBClassifier() + model.load_model(MODEL_FILE) + term_weights = joblib.load(TERM_WEIGHTS_FILE) + logger.info("Machine-Learning-Modell und Wortgewichte erfolgreich geladen.") except Exception as e: - logger.critical(f"Init GoogleSheetHandler fehlgeschlagen: {e}") - update_status(job_id, "Fehlgeschlagen", f"Init GoogleSheetHandler fehlgeschlagen: {e}") + logger.critical(f"Konnte Modelldateien nicht laden. Stelle sicher, dass '{MODEL_FILE}' und '{TERM_WEIGHTS_FILE}' vorhanden sind. Fehler: {e}") + update_status(job_id, "Fehlgeschlagen", f"Modelldateien nicht gefunden: {e}") + sys.exit(1) + + # Daten laden und vorbereiten + try: + sheet = GoogleSheetHandler() + crm_df = sheet.get_sheet_as_dataframe(CRM_SHEET_NAME) + match_df = sheet.get_sheet_as_dataframe(MATCHING_SHEET_NAME) + except Exception as e: + logger.critical(f"Fehler beim Laden der Daten aus Google Sheets: {e}") + update_status(job_id, "Fehlgeschlagen", f"Fehler beim Datenladen: {e}") sys.exit(1) - update_status(job_id, "Läuft", "Lade CRM- und Matching-Daten...") - crm_df = sheet.get_sheet_as_dataframe(CRM_SHEET_NAME) - match_df = sheet.get_sheet_as_dataframe(MATCHING_SHEET_NAME) total = len(match_df) if match_df is not None else 0 if crm_df is None or crm_df.empty or match_df is None or match_df.empty: logger.critical("Leere Daten in einem der Sheets. Abbruch.") @@ -203,7 +187,6 @@ def main(job_id=None, interactive=False): return logger.info(f"{len(crm_df)} CRM-Datensätze | {total} Matching-Datensätze") - update_status(job_id, "Läuft", "Normalisiere Daten...") for df in [crm_df, match_df]: df['normalized_name'] = df['CRM Name'].astype(str).apply(normalize_company_name) df['normalized_domain'] = df['CRM Website'].astype(str).apply(simple_normalize_url) @@ -212,23 +195,20 @@ def main(job_id=None, interactive=False): global CITY_TOKENS CITY_TOKENS = {t for s in pd.concat([crm_df['CRM Ort'], match_df['CRM Ort']]).dropna().unique() for t in _tokenize(s) if len(t) >= 3} - logger.info(f"City tokens gesammelt: {len(CITY_TOKENS)}") - term_weights = build_term_weights(crm_df) crm_records, domain_index, token_index = build_indexes(crm_df) - logger.info(f"Blocking: Domains={len(domain_index)} | TokenKeys={len(token_index)}") - - results = [] - logger.info("Starte Matching-Prozess…") + results = [] + logger.info("Starte Matching-Prozess mit ML-Modell…") + for idx, mrow in match_df.to_dict('index').items(): processed = idx + 1 progress_message = f"Prüfe {processed}/{total}: '{mrow.get('CRM Name','')}'" logger.info(progress_message) if processed % 10 == 0 or processed == total: update_status(job_id, "Läuft", progress_message) + # Kandidatensuche (Blocking) candidate_indices = set() - # ... (Kandidatensuche bleibt gleich) if mrow.get('normalized_domain'): candidates_from_domain = domain_index.get(mrow['normalized_domain'], []) for c in candidates_from_domain: @@ -241,7 +221,7 @@ def main(job_id=None, interactive=False): if not candidate_indices and rarest_token_mrec: candidate_indices.update(token_index.get(rarest_token_mrec, [])) - if not candidate_indices: + if len(candidate_indices) < 5: # Prefilter, wenn zu wenige Kandidaten pf = sorted([(fuzz.partial_ratio(clean_name_for_scoring(mrow.get('normalized_name',''))[0], clean_name_for_scoring(r.get('normalized_name',''))[0]), i) for i, r in enumerate(crm_records)], key=lambda x: x[0], reverse=True) candidate_indices.update([i for score, i in pf if score >= PREFILTER_MIN_PARTIAL][:PREFILTER_LIMIT]) @@ -250,50 +230,46 @@ def main(job_id=None, interactive=False): results.append({'Match':'', 'Score':0, 'Match_Grund':'keine Kandidaten'}) continue - scored = sorted([{'score': s, 'comp': c, 'record': r} for r in candidates for s, c in [calculate_similarity(mrow, r, term_weights, rarest_token_mrec)]], key=lambda x: x['score'], reverse=True) + # --- NEU: Prediction mit ML-Modell --- + feature_list = [] + for cr in candidates: + features = create_features(mrow, cr, term_weights) + feature_list.append(features) + + feature_df = pd.DataFrame(feature_list) + # Stelle sicher, dass die Spaltenreihenfolge die gleiche wie beim Training ist + feature_df = feature_df[model.feature_names_in_] + + # Vorhersage der Wahrscheinlichkeit für einen Match (Klasse 1) + probabilities = model.predict_proba(feature_df)[:, 1] - for cand in scored[:5]: logger.debug(f" Kandidat: {cand['record']['CRM Name']} | Score={cand['score']} | Comp={cand['comp']}") - best_match = scored[0] if scored else None + scored_candidates = [] + for i, prob in enumerate(probabilities): + scored_candidates.append({ + 'name': candidates[i].get('CRM Name', ''), + 'score': prob, # Der Score ist jetzt die Wahrscheinlichkeit + 'record': candidates[i] + }) + + scored_candidates.sort(key=lambda x: x['score'], reverse=True) - # --- Stufenmodell-Logik v5.0 --- - if best_match: - # Stufe 1 ist bereits in calculate_similarity behandelt (score=300) - - # Stufe 2: Intelligenter Tie-Breaker für Konzern-Logik - best_score = best_match['score'] - if len(scored) > 1 and best_score >= TRIGGER_SCORE_MIN and (best_score - scored[1]['score']) < TIE_SCORE_DIFF and best_score < GOLDEN_MATCH_SCORE: - if interactive: # Stufe 4: Manuelle Klärung - # ... (Interaktive Logik wie gehabt) - pass - else: # Stufe 2 Automatik - logger.info(f" Tie-Breaker-Situation erkannt. Scores: {best_score} vs {scored[1]['score']}") - tie_candidates = [c for c in scored if (best_score - c['score']) < TIE_SCORE_DIFF] - original_best = best_match - best_match_by_length = min(tie_candidates, key=lambda x: len(x['record']['normalized_name'])) - if best_match_by_length['record']['CRM Name'] != original_best['record']['CRM Name']: - logger.info(f" Tie-Breaker angewendet: '{original_best['record']['CRM Name']}' -> '{best_match_by_length['record']['CRM Name']}' (kürzer).") - best_match = best_match_by_length - - # Finale Entscheidung und Logging - if best_match and best_match['score'] >= SCORE_THRESHOLD: - is_weak = best_match['comp'].get('domain', 0) == 0 and not best_match['comp'].get('location', 0) - applied_threshold = SCORE_THRESHOLD_WEAK if is_weak else SCORE_THRESHOLD - if best_match['score'] >= applied_threshold: - results.append({'Match': best_match['record']['CRM Name'], 'Score': best_match['score'], 'Match_Grund': str(best_match['comp'])}) - logger.info(f" --> Match: '{best_match['record']['CRM Name']}' ({best_match['score']})") - else: - results.append({'Match':'', 'Score': best_match['score'], 'Match_Grund': f"Below WEAK TH | {str(best_match['comp'])}"}) - logger.info(f" --> No Match (below weak TH): '{best_match['record']['CRM Name']}' ({best_match['score']})") - elif best_match: - results.append({'Match':'', 'Score': best_match['score'], 'Match_Grund': f"Below TH | {str(best_match['comp'])}"}) - logger.info(f" --> No Match (below TH): '{best_match['record']['CRM Name']}' ({best_match['score']})") + best_match = scored_candidates[0] if scored_candidates else None + + # Finale Entscheidung basierend auf Threshold + if best_match and best_match['score'] >= PREDICTION_THRESHOLD: + results.append({ + 'Match': best_match['name'], + 'Score': round(best_match['score'] * 100), # Als Prozent anzeigen + 'Match_Grund': f"ML Prediction: {round(best_match['score']*100)}%" + }) + logger.info(f" --> Match: '{best_match['name']}' (Confidence: {round(best_match['score']*100)}%)") else: - results.append({'Match':'', 'Score':0, 'Match_Grund':'No valid candidates'}) - logger.info(f" --> No Match (no candidates)") + score_val = round(best_match['score'] * 100) if best_match else 0 + results.append({'Match':'', 'Score': score_val, 'Match_Grund': f"Below Threshold ({PREDICTION_THRESHOLD*100}%)"}) + logger.info(f" --> No Match (Confidence: {score_val}%)") - # --- Ergebnisse zurückschreiben --- + # Ergebnisse zurückschreiben logger.info("Matching-Prozess abgeschlossen. Schreibe Ergebnisse...") - update_status(job_id, "Läuft", "Schreibe Ergebnisse zurück ins Sheet...") result_df = pd.DataFrame(results) final_df = pd.concat([match_df.reset_index(drop=True), result_df.reset_index(drop=True)], axis=1) cols_to_drop = ['normalized_name', 'normalized_domain'] @@ -304,16 +280,17 @@ def main(job_id=None, interactive=False): ok = sheet.clear_and_write_data(MATCHING_SHEET_NAME, data_to_write) if ok: logger.info("Ergebnisse erfolgreich in das Google Sheet geschrieben.") - update_status(job_id, "Abgeschlossen", f"{total} Accounts erfolgreich geprüft.") + if job_id: update_status(job_id, "Abgeschlossen", f"{total} Accounts erfolgreich geprüft.") else: logger.error("Fehler beim Schreiben der Ergebnisse ins Google Sheet.") - update_status(job_id, "Fehlgeschlagen", "Fehler beim Schreiben ins Google Sheet.") + if job_id: update_status(job_id, "Fehlgeschlagen", "Fehler beim Schreiben ins Google Sheet.") if __name__=='__main__': - parser = argparse.ArgumentParser(description="Duplicate Checker v5.0") + parser = argparse.ArgumentParser(description="Duplicate Checker v5.0 (ML Model)") parser.add_argument("--job-id", type=str, help="Eindeutige ID für den Job-Status.") - parser.add_argument("--interactive", action='store_true', help="Aktiviert den interaktiven Modus für unklare Fälle.") args = parser.parse_args() + # Lade API-Keys etc. Config.load_api_keys() - main(job_id=args.job_id, interactive=args.interactive) \ No newline at end of file + + main(job_id=args.job_id) \ No newline at end of file