| from datetime import datetime, timedelta |
| from sqlmodel import select |
| from .models import Visit, TokenBalance |
| from .db import get_session |
|
|
| |
| W_RECENCY = 0.5 |
| W_FREQUENCY = 0.3 |
| W_POINTS = 0.2 |
|
|
| def calc_score(user_id: int) -> float: |
| now = datetime.utcnow() |
| with get_session() as s: |
| visits = s.exec( |
| select(Visit).where(Visit.user_id == user_id).order_by(Visit.created_at.desc()) |
| ).all() |
| if not visits: |
| return 0.0 |
|
|
| last_visit = visits[0].created_at |
| days_since = (now - last_visit).days |
|
|
| |
| recency_score = max(0.0, 1.0 - days_since / 30.0) |
|
|
| |
| recent_90 = now - timedelta(days=90) |
| freq_score = min( |
| 1.0, |
| len([v for v in visits if v.created_at > recent_90]) / 10.0 |
| ) |
|
|
| |
| balances = s.exec(select(TokenBalance).where(TokenBalance.user_id == user_id)).all() |
| points_amount = sum(tb.amount for tb in balances) |
| points_score = min(1.0, points_amount / 1000.0) |
|
|
| return W_RECENCY * recency_score + W_FREQUENCY * freq_score + W_POINTS * points_score |
|
|
| def segment(score: float) -> str: |
| if score >= 0.75: |
| return "loyal" |
| if score >= 0.4: |
| return "active" |
| if score > 0.0: |
| return "at-risk" |
| return "new" |
|
|
