1
from mlxtend.frequent_patterns import apriori, association_rules
2
from mlxtend.preprocessing import TransactionEncoder
3
import pandas as pd
4
import pickle
5
from datetime import datetime
6
7
class AssociationRuleRecommender:
8
"""Production-ready recommendation system using association rules."""
9
10
def __init__(self, min_support=0.02, min_confidence=0.6, min_lift=1.5):
11
self.min_support = min_support
12
self.min_confidence = min_confidence
13
self.min_lift = min_lift
14
self.rules = None
15
self.items_list = None
16
17
def train(self, transactions):
18
"""Train the recommender on historical transaction data."""
19
print(f"Training on {len(transactions)} transactions...")
20
21
# Encode transactions
22
te = TransactionEncoder()
23
te_array = te.fit(transactions).transform(transactions)
24
df = pd.DataFrame(te_array, columns=te.columns_)
25
self.items_list = list(te.columns_)
26
27
# Mine rules
28
frequent = apriori(df, min_support=self.min_support, use_colnames=True)
29
self.rules = association_rules(frequent, metric="confidence",
30
min_threshold=self.min_confidence)
31
self.rules = self.rules[self.rules['lift'] >= self.min_lift]
32
33
# Sort by lift for quick lookup
34
self.rules = self.rules.sort_values('lift', ascending=False)
35
36
print(f" Found {len(self.rules)} high-quality rules")
37
return self
38
39
def recommend(self, cart_items, top_n=5):
40
"""Generate recommendations based on current cart contents."""
41
if not self.rules or len(self.rules) == 0:
42
return []
43
44
cart_set = set(cart_items)
45
recommendations = []
46
47
# Find rules where all antecedents are in cart
48
for _, rule in self.rules.iterrows():
49
antecedents = set(rule['antecedents'])
50
consequents = set(rule['consequents'])
51
52
# Check if rule antecedents match cart items
53
if antecedents.issubset(cart_set):
54
# Recommend consequents not already in cart
55
new_items = consequents - cart_set
56
for item in new_items:
57
recommendations.append({
58
'item': item,
59
'confidence': rule['confidence'],
60
'lift': rule['lift'],
61
'reason': f"Often bought with {', '.join(antecedents)}"
62
})
63
64
# Deduplicate and rank by lift * confidence
65
seen = set()
66
unique_recs = []
67
for rec in recommendations:
68
if rec['item'] not in seen:
69
rec['score'] = rec['lift'] * rec['confidence']
70
unique_recs.append(rec)
71
seen.add(rec['item'])
72
73
# Return top N recommendations
74
return sorted(unique_recs, key=lambda x: x['score'], reverse=True)[:top_n]
75
76
def save(self, filepath):
77
"""Save trained model to disk."""
78
with open(filepath, 'wb') as f:
79
pickle.dump({
80
'rules': self.rules,
81
'items_list': self.items_list,
82
'params': {
83
'min_support': self.min_support,
84
'min_confidence': self.min_confidence,
85
'min_lift': self.min_lift
86
}
87
}, f)
88
print(f"Model saved to {filepath}")
89
90
@classmethod
91
def load(cls, filepath):
92
"""Load trained model from disk."""
93
with open(filepath, 'rb') as f:
94
data = pickle.load(f)
95
model = cls(**data['params'])
96
model.rules = data['rules']
97
model.items_list = data['items_list']
98
print(f"Model loaded from {filepath}")
99
return model
100
101
102
# EXAMPLE USAGE
103
# Training phase (offline, nightly batch job)
104
historical_transactions = [
105
['laptop', 'mouse', 'keyboard'],
106
['phone', 'charger', 'case'],
107
['laptop', 'mouse', 'monitor'],
108
['tablet', 'keyboard', 'stylus'],
109
['laptop', 'mouse', 'keyboard', 'usb_drive'],
110
] * 100 # 500 transactions
111
112
recommender = AssociationRuleRecommender(
113
min_support=0.05,
114
min_confidence=0.6,
115
min_lift=1.2
116
)
117
recommender.train(historical_transactions)
118
recommender.save('recommender_model.pkl')
119
120
# Production phase (real-time API)
121
# Load model once at server startup
122
loaded_recommender = AssociationRuleRecommender.load('recommender_model.pkl')
123
124
# User adds items to cart - generate recommendations
125
customer_cart = ['laptop', 'monitor']
126
recommendations = loaded_recommender.recommend(customer_cart, top_n=3)
127
128
print(f"\nCustomer cart: {customer_cart}")
129
print(f"Recommendations:")
130
for i, rec in enumerate(recommendations, 1):
131
print(f" {i}. {rec['item']}")
132
print(f" Confidence: {rec['confidence']:.0%} | Lift: {rec['lift']:.2f}")
133
print(f" Reason: {rec['reason']}")
134
135
# Output:
136
# Customer cart: ['laptop', 'monitor']
137
# Recommendations:
138
# 1. mouse
139
# Confidence: 90% | Lift: 1.80
140
# Reason: Often bought with laptop
141
# 2. keyboard
142
# Confidence: 75% | Lift: 1.50
143
# Reason: Often bought with laptop
144
# 3. usb_drive
145
# Confidence: 60% | Lift: 1.35
146
# Reason: Often bought with laptop, monitor
147
148
print(f"\nProduction-ready! Deploy as REST API or embed in e-commerce platform.")
