이 글은 누구를 위한 것인가
- 모든 고객에게 동일한 메시지를 보내는 비효율적인 마케팅에서 벗어나고 싶은 팀
- 고객 데이터는 있는데 어떻게 활용할지 모르는 이커머스 데이터 분석가
- RFM을 들어봤지만 실제 구현 방법이 필요한 개발자·PM
들어가며
이커머스의 80/20 법칙: 매출의 80%는 상위 20% 고객에서 나온다. 그렇다면 이 20%는 누구인가? RFM은 가장 검증된 방법이다.
RFM = Recency(최근 구매일), Frequency(구매 횟수), Monetary(구매 금액). 이 세 지표만으로도 고객을 의미 있게 분류할 수 있고, 세그먼트별로 다른 마케팅을 적용하면 CRM 효율이 극적으로 높아진다.
이 글은 bluefoxdev.kr의 이커머스 데이터 분석 가이드 를 참고하고, RFM 실전 구현 관점에서 확장하여 작성했습니다.
1. RFM 스코어링 구현
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
def calculate_rfm(orders_df: pd.DataFrame, analysis_date: datetime) -> pd.DataFrame:
"""
orders_df 컬럼: customer_id, order_date, order_amount
"""
rfm = orders_df.groupby('customer_id').agg({
'order_date': lambda x: (analysis_date - x.max()).days, # Recency
'order_id': 'count', # Frequency
'order_amount': 'sum' # Monetary
}).reset_index()
rfm.columns = ['customer_id', 'recency', 'frequency', 'monetary']
# 각 지표를 1~5 점수로 변환 (분위수 기반)
# Recency: 낮을수록(최근) 높은 점수
rfm['r_score'] = pd.qcut(rfm['recency'], 5, labels=[5, 4, 3, 2, 1]).astype(int)
# Frequency: 높을수록 높은 점수
rfm['f_score'] = pd.qcut(rfm['frequency'].rank(method='first'), 5, labels=[1, 2, 3, 4, 5]).astype(int)
# Monetary: 높을수록 높은 점수
rfm['m_score'] = pd.qcut(rfm['monetary'].rank(method='first'), 5, labels=[1, 2, 3, 4, 5]).astype(int)
# 종합 RFM 점수 (문자열 조합)
rfm['rfm_score'] = rfm['r_score'].astype(str) + rfm['f_score'].astype(str) + rfm['m_score'].astype(str)
# RFM 합산 점수
rfm['rfm_total'] = rfm['r_score'] + rfm['f_score'] + rfm['m_score']
return rfm
2. 고객 세그먼트 분류
def assign_segment(rfm_df: pd.DataFrame) -> pd.DataFrame:
"""RFM 패턴에 따른 세그먼트 자동 분류"""
def segment_customer(row):
r, f, m = row['r_score'], row['f_score'], row['m_score']
rfm = row['rfm_score']
# 챔피언: 최근 구매, 자주 구매, 많이 구매
if r >= 4 and f >= 4 and m >= 4:
return '챔피언'
# 충성 고객: 자주 구매하며 꾸준한 소비
elif r >= 3 and f >= 4:
return '충성 고객'
# 잠재 충성 고객: 최근 구매, 여러 번 구매
elif r >= 4 and f >= 2:
return '잠재 충성 고객'
# 신규 고객: 최근 구매했지만 구매 횟수 적음
elif r >= 4 and f == 1:
return '신규 고객'
# 위험 고객: 예전엔 좋았는데 최근에 안 옴
elif r <= 2 and f >= 3 and m >= 3:
return '위험 고객'
# 잠자는 고객: 오래됐고 빈도도 낮음
elif r <= 2 and f <= 2:
return '이탈 위험'
# 큰손: 많이 쓰지만 가끔 옴
elif m >= 4 and f <= 2:
return '큰손'
else:
return '일반'
rfm_df['segment'] = rfm_df.apply(segment_customer, axis=1)
return rfm_df
3. K-Means 클러스터링으로 자동 세분화
from sklearn.preprocessing import StandardScaler
from sklearn.cluster import KMeans
from sklearn.metrics import silhouette_score
import matplotlib.pyplot as plt
def auto_segment_kmeans(rfm_df: pd.DataFrame, n_clusters: int = 5) -> pd.DataFrame:
"""K-Means로 데이터 기반 자동 클러스터링"""
features = rfm_df[['recency', 'frequency', 'monetary']].copy()
# 로그 변환 (왜도 제거)
features['log_recency'] = np.log1p(features['recency'])
features['log_frequency'] = np.log1p(features['frequency'])
features['log_monetary'] = np.log1p(features['monetary'])
# 정규화
scaler = StandardScaler()
scaled = scaler.fit_transform(
features[['log_recency', 'log_frequency', 'log_monetary']]
)
# 최적 클러스터 수 찾기 (Elbow Method)
inertias = []
silhouettes = []
k_range = range(3, 9)
for k in k_range:
km = KMeans(n_clusters=k, random_state=42, n_init=10)
labels = km.fit_predict(scaled)
inertias.append(km.inertia_)
silhouettes.append(silhouette_score(scaled, labels))
# 최적 k 선택 (Silhouette Score 최대)
best_k = k_range[silhouettes.index(max(silhouettes))]
print(f"최적 클러스터 수: {best_k} (Silhouette: {max(silhouettes):.3f})")
# 최종 클러스터링
km_final = KMeans(n_clusters=best_k, random_state=42, n_init=10)
rfm_df['cluster'] = km_final.fit_predict(scaled)
# 클러스터별 특성 분석
cluster_profile = rfm_df.groupby('cluster').agg({
'recency': 'mean',
'frequency': 'mean',
'monetary': 'mean',
'customer_id': 'count'
}).round(1)
print("\n클러스터 프로파일:")
print(cluster_profile)
return rfm_df, cluster_profile
4. 세그먼트별 마케팅 자동화
SEGMENT_STRATEGIES = {
'챔피언': {
'action': 'vip_reward',
'message': '당신은 우리의 VIP입니다. 특별 혜택을 드립니다.',
'discount': 0, # 할인 불필요
'channel': ['push', 'email'],
'frequency': 'monthly', # 과잉 접촉 방지
},
'충성 고객': {
'action': 'loyalty_upsell',
'message': '단골 고객 감사 혜택',
'discount': 5,
'channel': ['push', 'email'],
'frequency': 'biweekly',
},
'위험 고객': {
'action': 'winback_offer',
'message': '보고 싶었어요! 돌아오시면 특별 할인을',
'discount': 15,
'channel': ['email', 'sms'],
'frequency': 'once', # 한 번만 (스팸 방지)
'urgency': '7일 한정',
},
'이탈 위험': {
'action': 'last_chance_winback',
'message': '마지막 특별 제안',
'discount': 20,
'channel': ['email'],
'frequency': 'once',
},
'신규 고객': {
'action': 'nurture_onboarding',
'message': '첫 구매 감사합니다! 다음 구매 혜택',
'discount': 10,
'channel': ['email', 'push'],
'frequency': 'weekly',
},
'큰손': {
'action': 'category_recommend',
'message': '취향에 맞는 프리미엄 신상품',
'discount': 0, # 가격 민감도 낮음
'channel': ['email'],
'frequency': 'monthly',
},
}
def generate_campaign_list(rfm_df: pd.DataFrame) -> pd.DataFrame:
"""세그먼트별 캠페인 대상자 추출"""
campaigns = []
for segment, strategy in SEGMENT_STRATEGIES.items():
customers = rfm_df[rfm_df['segment'] == segment]['customer_id'].tolist()
if not customers:
continue
campaigns.append({
'segment': segment,
'customer_count': len(customers),
'action': strategy['action'],
'discount': strategy['discount'],
'channels': ', '.join(strategy['channel']),
'expected_roi': estimate_roi(segment, len(customers), strategy['discount']),
})
print(f"[{segment}] {len(customers)}명 → {strategy['action']} ({strategy['discount']}% 할인)")
return pd.DataFrame(campaigns)
def estimate_roi(segment: str, count: int, discount: float) -> str:
"""세그먼트별 예상 ROI 계산"""
base_cvr = {
'챔피언': 0.35, '충성 고객': 0.25, '위험 고객': 0.10,
'이탈 위험': 0.05, '신규 고객': 0.20, '큰손': 0.15
}
cvr = base_cvr.get(segment, 0.10)
expected_orders = count * cvr
return f"전환 예상 {expected_orders:.0f}건"
5. 자동화 스케줄링
# Celery + Redis로 주간 자동 실행
from celery import Celery
from celery.schedules import crontab
app = Celery('rfm_tasks', broker='redis://localhost:6379/0')
@app.task
def run_weekly_rfm_analysis():
"""매주 월요일 오전 2시 자동 실행"""
from datetime import datetime
import sqlalchemy as sa
engine = sa.create_engine(DATABASE_URL)
# 최근 2년 주문 데이터 로드
orders_df = pd.read_sql("""
SELECT customer_id, order_id, order_date, total_amount as order_amount
FROM orders
WHERE order_date >= NOW() - INTERVAL '2 years'
AND status = 'completed'
""", engine)
analysis_date = datetime.now()
# RFM 계산
rfm = calculate_rfm(orders_df, analysis_date)
rfm = assign_segment(rfm)
# DB 저장
rfm.to_sql('customer_segments', engine, if_exists='replace', index=False)
# 캠페인 생성
campaigns = generate_campaign_list(rfm)
# 마케팅 자동화 시스템에 전달
for _, campaign in campaigns.iterrows():
send_to_marketing_automation(campaign)
print(f"RFM 분석 완료: {len(rfm)}명 분류")
return len(rfm)
# 스케줄 설정
app.conf.beat_schedule = {
'weekly-rfm': {
'task': 'tasks.run_weekly_rfm_analysis',
'schedule': crontab(hour=2, minute=0, day_of_week=1), # 매주 월 2시
},
}
마무리: RFM 적용 효과
실제 적용 사례 (이커머스 기준):
세그먼트별 캠페인 효율 비교:
┌──────────────┬──────────┬──────────┬──────────┐
│ 세그먼트 │ 기존 CVR │ RFM CVR │ 개선율 │
├──────────────┼──────────┼──────────┼──────────┤
│ 챔피언 │ 5.2% │ 28.4% │ +446% │
│ 위험 고객 │ 5.2% │ 12.1% │ +133% │
│ 이탈 위험 │ 5.2% │ 4.8% │ -8% │
└──────────────┴──────────┴──────────┴──────────┘
→ 전체 발송량 40% 감소, 전환 수익 35% 증가
RFM은 복잡한 ML 없이도 즉시 적용 가능하다. 규칙 기반 세분화로 시작하고, 데이터가 쌓이면 K-Means로 자동화하는 것이 현실적인 로드맵이다.