Multi-Stage Fuzzy Matching
Production-grade fuzzy matching for inventory items, products, or entity reconciliation.
When to Use This Skill
- Matching vendor SKUs to inventory items
- Entity reconciliation with varying name formats
- Product deduplication across sources
- Any scenario where exact matching misses valid matches
Core Concepts
Three-stage pipeline: PostgreSQL trigram (fast pre-filter) → Salient overlap (quick filter) → Multi-factor similarity (expensive, accurate). Achieves O(log n) with proper indexing.
Stage 1: PostgreSQL Trigram (fast) → 50 candidates
Stage 2: Salient Overlap Check (fast) → ~20 candidates
Stage 3: Multi-Factor Similarity (expensive) → ranked resultsImplementation
Python
import re
import math
from typing import List, Dict, Optional, Set
from decimal import Decimal
class TextNormalizer:
"""Normalizes text for consistent matching."""
BRAND_PATTERNS = [r'\bsysco\b', r'\bus foods\b', r'\bpremium\b', r'\bselect\b']
UNIT_MAP = {
'lb': 'pound', 'lbs': 'pound', 'oz': 'ounce',
'kg': 'kilogram', 'g': 'gram', 'gal': 'gallon',
}
STOPWORDS = {'the', 'and', 'or', 'with', 'of', 'boneless', 'bnls', 'fresh', 'frozen'}
def normalize_text(self, text: str) -> str:
if not text:
return ""
normalized = text.lower().strip()
for pattern in self.BRAND_PATTERNS:
normalized = re.sub(pattern, '', normalized, flags=re.IGNORECASE)
for variant, standard in self.UNIT_MAP.items():
normalized = re.sub(r'\b' + variant + r'\b', standard, normalized, flags=re.IGNORECASE)
normalized = re.sub(r'[^\w\s-]', ' ', normalized)
return ' '.join(normalized.split())
def tokenize(self, text: str) -> List[str]:
normalized = self.normalize_text(text)
tokens = re.split(r'[\s-]+', normalized)
return [t for t in tokens if t and t not in self.STOPWORDS and len(t) >= 2]
class SimilarityCalculator:
"""Calculates multi-factor similarity scores."""
WEIGHTS = {
'name_similarity': 0.55,
'token_similarity': 0.25,
'size_similarity': 0.15,
'category_similarity': 0.05,
}
def has_salient_overlap(self, tokens1: List[str], tokens2: List[str]) -> bool:
"""Quick pre-filter: do items share any salient words?"""
salient1 = {t for t in tokens1 if len(t) >= 3}
salient2 = {t for t in tokens2 if len(t) >= 3}
if not salient1 or not salient2:
return False
return len(salient1 & salient2) > 0
def trigram_cosine_similarity(self, text1: str, text2: str) -> float:
if not text1 or not text2:
return 0.0
if text1 == text2:
return 1.0
def get_trigrams(text: str) -> Set[str]:
padded = f" {text} "
return {padded[i:i+3] for i in range(len(padded) - 2)}
t1, t2 = get_trigrams(text1), get_trigrams(text2)
intersection = t1 & t2
if not intersection:
return 0.0
return len(intersection) / math.sqrt(len(t1) * len(t2))
def weighted_jaccard_similarity(self, tokens1: List[str], tokens2: List[str]) -> float:
if not tokens1 or not tokens2:
return 0.0
set1, set2 = set(tokens1), set(tokens2)
if set1 == set2:
return 1.0
def weight(t: str) -> float:
return 2.0 if len(t) >= 5 else 1.5 if len(t) >= 3 else 1.0
intersection = set1 & set2
union = set1 | set2
return sum(weight(t) for t in intersection) / sum(weight(t) for t in union)
def size_similarity(self, size1: Optional[Decimal], size2: Optional[Decimal]) -> float:
if size1 is None or size2 is None:
return 0.5
if size1 == size2:
return 1.0
ratio = float(min(size1, size2) / max(size1, size2))
if ratio >= 0.95: return 1.0
if ratio >= 0.85: return 0.8
if ratio >= 0.70: return 0.5
return 0.0
def calculate_advanced_similarity(self, item1: Dict, item2: Dict) -> float:
name1 = item1.get('normalized_name', '')
name2 = item2.get('normalized_name', '')
name_sim = self.trigram_cosine_similarity(name1, name2)
token_sim = self.weighted_jaccard_similarity(
self.normalizer.tokenize(name1),
self.normalizer.tokenize(name2)
)
size_sim = self.size_similarity(
self.extract_size(name1),
self.extract_size(name2)
)
cat_sim = 1.0 if item1.get('category') == item2.get('category') else 0.0
return (
self.WEIGHTS['name_similarity'] * name_sim +
self.WEIGHTS['token_similarity'] * token_sim +
self.WEIGHTS['size_similarity'] * size_sim +
self.WEIGHTS['category_similarity'] * cat_sim
)class FuzzyItemMatcher:
"""Complete 3-stage fuzzy matching pipeline."""
THRESHOLDS = {
'auto_match': 0.95,
'review_match': 0.85,
'min_similarity': 0.70,
'trigram_filter': 0.3,
}
def __init__(self, db_client):
self.client = db_client
self.normalizer = TextNormalizer()
self.calculator = SimilarityCalculator()
def find_similar_items(
self,
target_name: str,
user_id: str,
category: Optional[str] = None,
limit: int = 10,
) -> List[Dict]:
normalized_target = self.normalizer.normalize_text(target_name)
target_tokens = self.normalizer.tokenize(target_name)
# Stage 1: PostgreSQL trigram (fast)
candidates = self._trigram_search(normalized_target, user_id, category, 0.3, 50)
if not candidates:
return []
# Stage 2: Salient overlap filter (fast)
filtered = []
for candidate in candidates:
candidate_tokens = self.normalizer.tokenize(candidate['normalized_name'])
if self.calculator.has_salient_overlap(target_tokens, candidate_tokens):
filtered.append(candidate)
if not filtered:
return []
# Stage 3: Advanced similarity (expensive)
target_item = {'normalized_name': normalized_target, 'category': category}
results = []
for candidate in filtered:
similarity = self.calculator.calculate_advanced_similarity(target_item, candidate)
if similarity >= self.THRESHOLDS['min_similarity']:
results.append({**candidate, 'similarity_score': similarity})
results.sort(key=lambda x: x['similarity_score'], reverse=True)
return results[:limit]
def _trigram_search(self, name: str, user_id: str, category: Optional[str], threshold: float, limit: int) -> List[Dict]:
result = self.client.rpc('find_similar_items', {
'target_name': name,
'target_user_id': user_id,
'similarity_threshold': threshold,
'result_limit': limit,
}).execute()
return result.data or []
def get_match_recommendation(self, score: float) -> Dict:
if score >= self.THRESHOLDS['auto_match']:
return {'action': 'auto_match', 'confidence': 'high', 'needs_review': False}
elif score >= self.THRESHOLDS['review_match']:
return {'action': 'review', 'confidence': 'medium', 'needs_review': True}
return {'action': 'create_new', 'confidence': 'low', 'needs_review': False}SQL Setup
CREATE EXTENSION IF NOT EXISTS pg_trgm;
ALTER TABLE inventory_items ADD COLUMN normalized_name TEXT;
CREATE INDEX idx_items_trgm ON inventory_items USING GIN (normalized_name gin_trgm_ops);
CREATE FUNCTION find_similar_items(target_name TEXT, target_user_id UUID, similarity_threshold FLOAT, result_limit INT)
RETURNS TABLE (id UUID, name TEXT, normalized_name TEXT, category TEXT, similarity_score FLOAT) AS $$
BEGIN
RETURN QUERY
SELECT i.id, i.name, i.normalized_name, i.category,
similarity(i.normalized_name, target_name) as similarity_score
FROM inventory_items i
WHERE i.user_id = target_user_id
AND similarity(i.normalized_name, target_name) > similarity_threshold
ORDER BY similarity_score DESC
LIMIT result_limit;
END;
$$ LANGUAGE plpgsql;Usage Examples
matcher = FuzzyItemMatcher(db_client)
# Find matches for vendor SKU
matches = matcher.find_similar_items(
target_name="BNLS CHKN BRST 10LB",
user_id="user_123",
category="poultry",
)
for match in matches:
rec = matcher.get_match_recommendation(match['similarity_score'])
print(f"{match['name']}: {match['similarity_score']:.2f} - {rec['action']}")Best Practices
- Normalize on write - store normalized_name column
- Start with conservative thresholds (0.95 auto-match)
- Use domain-specific stopwords (industry terms)
- Always have Python fallback if RPC fails
- Tune weights based on your data characteristics
Common Mistakes
- Computing normalization on every query (slow)
- Generic stopword lists missing domain terms
- Not handling size/unit variations
- Skipping the salient overlap filter (slow)
Related Patterns
- deduplication (preventing duplicates)
- validation-quarantine (handling uncertain matches)