リードスコアリング完全ガイド
7つの手法と実装パターン

SELECT
  c.contact_id,
  c.company_name,
  c.email,
  -- 属性スコア
  CASE
    WHEN c.employee_count >= 500 THEN 15
    WHEN c.employee_count >= 100 THEN 10
    WHEN c.employee_count >= 50  THEN 5
    ELSE 0
  END
  + CASE
      WHEN c.job_title LIKE '%部長%' OR c.job_title LIKE '%Director%' THEN 15
      WHEN c.job_title LIKE '%課長%' OR c.job_title LIKE '%Manager%' THEN 10
      ELSE 5
    END
  + CASE WHEN c.industry IN ('SaaS', 'IT', '製造業') THEN 10 ELSE 0 END
  -- 行動スコア
  + COALESCE(e.seminar_count, 0) * 30
  + CASE WHEN e.pricing_page_views > 0 THEN 20 ELSE 0 END
  + COALESCE(e.downloads, 0) * 20
  -- 減点スコア
  - CASE WHEN c.email LIKE '%@competitor.com' THEN 1000 ELSE 0 END
  - CASE WHEN c.email LIKE '%@gmail.com' THEN 10 ELSE 0 END
  AS total_score
FROM contacts c
LEFT JOIN engagement_summary e ON c.contact_id = e.contact_id
WHERE c.lifecycle_stage = 'lead'
ORDER BY total_score DESC;

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import classification_report, roc_auc_score

# 特徴量設計: CRM + 行動データ
features = [
    'employee_count',        # 従業員数
    'is_target_industry',    # ターゲット業種フラグ
    'job_title_level',       # 役職レベル(1-5)
    'page_views_total',      # 総PV
    'pricing_page_views',    # 価格ページPV
    'email_open_rate',       # メール開封率
    'days_since_last_visit', # 最終訪問からの日数
    'content_downloads',     # 資料DL数
    'seminar_attendance',    # セミナー参加回数
]

# データ準備
df = pd.read_csv('leads_with_outcomes.csv')
X = df[features]
y = df['converted']  # 1=受注, 0=失注

# 前処理
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
X_train, X_test, y_train, y_test = train_test_split(
    X_scaled, y, test_size=0.2, random_state=42, stratify=y
)

# 学習
model = LogisticRegression(class_weight='balanced')
model.fit(X_train, y_train)

# 評価
y_pred = model.predict(X_test)
y_proba = model.predict_proba(X_test)[:, 1]
print(classification_report(y_test, y_pred))
print(f"AUC-ROC: {roc_auc_score(y_test, y_proba):.3f}")

import xgboost as xgb
from imblearn.over_sampling import SMOTE
from sklearn.model_selection import GridSearchCV
import shap

# 不均衡データ対策: SMOTEでオーバーサンプリング
smote = SMOTE(random_state=42)
X_resampled, y_resampled = smote.fit_resample(X_train, y_train)

# ハイパーパラメータ探索
param_grid = {
    'max_depth': [3, 5, 7],
    'learning_rate': [0.01, 0.1, 0.3],
    'n_estimators': [100, 300, 500],
    'min_child_weight': [1, 3, 5],
}

grid = GridSearchCV(
    xgb.XGBClassifier(eval_metric='logloss'),
    param_grid,
    cv=5,
    scoring='roc_auc',
    n_jobs=-1
)
grid.fit(X_resampled, y_resampled)

best_model = grid.best_estimator_
print(f"Best AUC: {grid.best_score_:.3f}")

# SHAP値で予測の根拠を可視化
explainer = shap.TreeExplainer(best_model)
shap_values = explainer.shap_values(X_test)
shap.summary_plot(shap_values, X_test, feature_names=features)

import anthropic
import json

client = anthropic.Anthropic()

icp_definition = """
理想顧客像（ICP）:
- 従業員数 50-500名の B2B SaaS企業
- シリーズA以降の資金調達済み
- CRM（HubSpot or Salesforce）を導入済み
- 営業チームが5名以上
- 直近6ヶ月以内に営業関連の求人を出している
- 日本市場で事業展開中
"""

def score_lead(lead_data: dict) -> dict:
    response = client.messages.create(
        model="claude-sonnet-4-5-20250514",
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"""
以下のICP定義に対して、このリードの適合度を0-100で評価せよ。
スコアと理由をJSON形式で返せ。

ICP定義:
{icp_definition}

リード情報:
{json.dumps(lead_data, ensure_ascii=False)}

出力形式:
{{"score": 数値, "fit_reason": "適合理由", "risk": "懸念点", "next_action": "推奨アクション"}}
"""
        }]
    )
    return json.loads(response.content[0].text)

# 使用例
lead = {
    "company": "株式会社Example",
    "employees": 120,
    "industry": "B2B SaaS",
    "funding": "Series B",
    "crm": "HubSpot",
    "sales_team_size": 8,
    "recent_job_posts": ["セールスマネージャー", "ISR"]
}
result = score_lead(lead)
print(f"Score: {result['score']} - {result['next_action']}")

from datetime import datetime, timedelta

def apply_decay(original_score: float, last_activity: datetime) -> float:
    """段階的減衰モデルでスコアを減衰させる"""
    days_elapsed = (datetime.now() - last_activity).days

    decay_rules = [
        (30,  1.0),   # 30日以内: 減衰なし
        (60,  0.9),   # 31-60日: -10%
        (90,  0.7),   # 61-90日: -30%
        (180, 0.5),   # 91-180日: -50%
        (365, 0.2),   # 181-365日: -80%
    ]

    for threshold, multiplier in decay_rules:
        if days_elapsed <= threshold:
            return round(original_score * multiplier, 1)

    return 0.0  # 365日超過: リセット

# バッチ処理: 全リードのスコアを日次で減衰適用
def batch_decay_scores(leads: list[dict]) -> list[dict]:
    return [
        {
            **lead,
            'decayed_score': apply_decay(
                lead['score'],
                lead['last_activity_at']
            )
        }
        for lead in leads
    ]