Silnik grywalizacji: architektura i maszyna stanu
80% uczniów twierdzi, że jest bardziej zmotywowana, gdy uczy się na takich platformach włączyć elementy gry. Rynek grywalizacji w EdTech będzie rósł w tempie ok CAGR powyżej 36% do 2032 r. Ale jaka jest różnica między grywalizacją skuteczne i powierzchowne nie polega na dodawaniu przypadkowych odznak i punktów: chodzi o solidność zpodstawowa architektura.
Źle zaprojektowany silnik grywalizacji szybko gromadzi dług techniczny: logikę zakodowany na stałe postęp, stan wyścigu na punktach, gdy wielu użytkowników ukończy działań w tym samym czasie, tabele wyników, które stają się wąskimi gardłami dla tysięcy użytkowników, odznaki przyznane kilkukrotnie ze względu na błędy w zarządzaniu statusami. Te problemy nie są teoretyczne: to typowe patologie każdej platformy, która dodaje grywalizację po namyśle, zamiast projektować go jako funkcję pierwszorzędną.
W tym artykule zbudujemy Silnik grywalizacji solidny i skalowalny z jawnymi maszynami stanu, pozyskiwaniem zdarzeń w celu śledzenia, Redis dla rankingów działający w czasie rzeczywistym i konfigurowalny system reguł, który umożliwia twórcom treści definiowanie nowa mechanika bez dotykania kodu.
Czego dowiesz się w tym artykule
- Sterowana zdarzeniami architektura skalowalnego silnika grywalizacji
- Maszyna stanu do zarządzania postępami uczniów
- System identyfikatorów z konfigurowalnymi regułami (bez kodowania)
- Tablica wyników w czasie rzeczywistym z zestawami sortowanymi Redis
- Śledzenie serii i system nagród czasowych
- XP (punkty doświadczenia) z poziomami i krzywymi postępu
- Anty-wzorce: jak uniknąć sytuacji wyścigu i podwójnego przypisania
- Powiadomienia push dotyczące osiągnięć dzięki WebSocket
1. Zasady projektowania: grywalizacja wewnętrzna i zewnętrzna
Przed napisaniem kodu konieczne jest zrozumienie różnicy między motywacją zewnętrzny (punkty, odznaki, rankingi – motywatory zewnętrzne) e wewnętrzny (ciekawość, poczucie kompetencji, autonomia). Skuteczna grywalizacja edukacyjna wykorzystuje motywatory zewnętrzne jako platformę startową rozwijać motywację wewnętrzną, a nie substytut.
Najbardziej efektywne mechanizmy grywalizacji w edukacji w 2025 roku to: pasemko (sekwencje kolejnych dni nauki), progresja mistrzostwa (wizualizacja postępów w kierunku mistrzostwa), uczenie się społeczne (rankingi wśród rówieśników o podobnym poziomie), tj sensowne wybory (alternatywne ścieżki z różnymi nagrodami). Najmniej skuteczną (i potencjalnie szkodliwą) mechaniką są globalne rankingi gdzie nowi studenci zawsze postrzegają najlepszych wykonawców jako nieosiągalnych.
Komponenty silnika grywalizacji
| Część | Odpowiedzialność | Technologia |
|---|---|---|
| Autobus imprezowy | Gromadzi zdarzenia edukacyjne | Strumienie Kafki/Redisa |
| Silnik reguł | Oceń konfigurowalne reguły | Python + reguły JSON |
| Maszyna Stanu | Zarządza postępami uczniów | Niestandardowy Python FSM |
| Kalkulator XP | Oblicz punkty doświadczenia | Python + PostgreSQL |
| Menedżer odznak | Przyznaj odznakę idempotentną | Pamięć podręczna PostgreSQL + Redis |
| Tabele liderów | Rankingi w czasie rzeczywistym | Sortowane zestawy Redis |
| Śledzenie smug | Monitoruj sekwencje badań | Redis + PostgreSQL |
| Centrum powiadomień | Push osiągnięcia w czasie rzeczywistym | WebSocket/SSE |
2. Maszyna stanu dla postępu uczniów
Maszyna stanów jest sercem silnika grywalizacji. Każdy uczeń jest w jednym państwo co odzwierciedla poziom jego zaangażowania i postępu. Przejścia między stanami zachodzą w odpowiedzi na wydarzenia (ukończenie lekcja, poprawna odpowiedź, kolejne logowanie). Uczyń tę maszynę stanu wyraźną eliminuje problem „stanów ukrytych”, które powodują trudne do odtworzenia błędy.
# gamification/state_machine.py
from enum import Enum, auto
from dataclasses import dataclass, field
from typing import Dict, List, Callable, Optional
from datetime import datetime
class LearnerState(Enum):
"""Stati possibili di un learner nella piattaforma."""
ONBOARDING = auto() # Primo accesso, < 3 corsi completati
ACTIVE = auto() # Studente attivo, streak in corso
AT_RISK = auto() # Nessuna attivita da 7-13 giorni
DORMANT = auto() # Inattivo da 14+ giorni
MASTERY = auto() # Ha completato un corso con punteggio >= 90%
MENTOR = auto() # Ha aiutato almeno 5 altri studenti
class LearnerEvent(Enum):
"""Eventi che causano transizioni di stato."""
LESSON_COMPLETED = "lesson_completed"
QUIZ_PASSED = "quiz_passed"
QUIZ_FAILED = "quiz_failed"
COURSE_COMPLETED = "course_completed"
PEER_HELPED = "peer_helped"
DAILY_LOGIN = "daily_login"
STREAK_BROKEN = "streak_broken"
INACTIVE_7_DAYS = "inactive_7_days"
INACTIVE_14_DAYS = "inactive_14_days"
REACTIVATED = "reactivated"
@dataclass
class StateTransition:
from_state: LearnerState
event: LearnerEvent
to_state: LearnerState
conditions: List[Callable] = field(default_factory=list)
actions: List[Callable] = field(default_factory=list)
@dataclass
class LearnerContext:
student_id: str
current_state: LearnerState
xp: int = 0
streak_days: int = 0
courses_completed: int = 0
peers_helped: int = 0
last_activity: Optional[datetime] = None
metadata: Dict = field(default_factory=dict)
class LearnerStateMachine:
"""
State machine per la progressione del learner.
Tutte le transizioni sono esplicite e testabili.
"""
TRANSITIONS: List[StateTransition] = [
# ONBOARDING -> ACTIVE: ha completato la prima lezione
StateTransition(
from_state=LearnerState.ONBOARDING,
event=LearnerEvent.LESSON_COMPLETED,
to_state=LearnerState.ACTIVE,
),
# ACTIVE -> AT_RISK: 7 giorni di inattivita
StateTransition(
from_state=LearnerState.ACTIVE,
event=LearnerEvent.INACTIVE_7_DAYS,
to_state=LearnerState.AT_RISK,
),
# AT_RISK -> DORMANT: altri 7 giorni senza attivita
StateTransition(
from_state=LearnerState.AT_RISK,
event=LearnerEvent.INACTIVE_14_DAYS,
to_state=LearnerState.DORMANT,
),
# DORMANT/AT_RISK -> ACTIVE: torna ad essere attivo
StateTransition(
from_state=LearnerState.AT_RISK,
event=LearnerEvent.REACTIVATED,
to_state=LearnerState.ACTIVE,
),
StateTransition(
from_state=LearnerState.DORMANT,
event=LearnerEvent.REACTIVATED,
to_state=LearnerState.ACTIVE,
),
# ACTIVE -> MASTERY: completa corso con voto alto
StateTransition(
from_state=LearnerState.ACTIVE,
event=LearnerEvent.COURSE_COMPLETED,
to_state=LearnerState.MASTERY,
conditions=[lambda ctx: ctx.metadata.get("score", 0) >= 90],
),
# MASTERY -> MENTOR: aiuta abbastanza peer
StateTransition(
from_state=LearnerState.MASTERY,
event=LearnerEvent.PEER_HELPED,
to_state=LearnerState.MENTOR,
conditions=[lambda ctx: ctx.peers_helped >= 5],
),
]
def __init__(self, event_publisher):
self.event_publisher = event_publisher
# Indice per lookup rapido O(1)
self._transition_index: Dict[tuple, StateTransition] = {
(t.from_state, t.event): t
for t in self.TRANSITIONS
}
def process_event(
self,
context: LearnerContext,
event: LearnerEvent,
event_data: Dict = None,
) -> LearnerContext:
"""
Processa un evento e restituisce un NUOVO context (immutabile).
Non modifica mai il context in input.
"""
event_data = event_data or {}
key = (context.current_state, event)
transition = self._transition_index.get(key)
if not transition:
# Nessuna transizione definita: evento ignorato
return context
# Aggiorna il metadata del context con i dati dell'evento
updated_metadata = {**context.metadata, **event_data}
temp_context = LearnerContext(
**{
**context.__dict__,
"metadata": updated_metadata,
}
)
# Verifica tutte le condizioni
if not all(cond(temp_context) for cond in transition.conditions):
return context # Condizioni non soddisfatte, nessuna transizione
# Crea nuovo context con nuovo stato
new_context = LearnerContext(
**{
**temp_context.__dict__,
"current_state": transition.to_state,
"last_activity": datetime.utcnow(),
}
)
# Pubblica evento di transizione
self.event_publisher.publish({
"type": "state_transition",
"student_id": context.student_id,
"from_state": context.current_state.name,
"to_state": transition.to_state.name,
"trigger_event": event.value,
"timestamp": datetime.utcnow().isoformat(),
})
return new_context3. System XP i poziomów z krzywymi postępu
System punktów doświadczenia (XP) musi równoważyć szybkość postępu: za szybko i uczniowie osiągają maksimum w ciągu kilku dni i nudzą się, zbyt wolno i zniechęcają się. Użyjmy jednego krzywa logarytmiczna przez kilka pierwszych poziomów (szybki i satysfakcjonujący postęp) robi się bardziej stromo ku wysokim poziomom (znaczące cele).
# gamification/xp_system.py
import math
from dataclasses import dataclass
from typing import Dict, List
from enum import Enum
class ActivityType(Enum):
LESSON_COMPLETED = "lesson"
QUIZ_PASSED = "quiz_pass"
QUIZ_PERFECT = "quiz_perfect" # 100% corretto
COURSE_COMPLETED = "course"
STREAK_BONUS = "streak"
PEER_REVIEW = "peer_review"
HELP_PEER = "help_peer"
FIRST_LOGIN_DAY = "first_login"
# XP base per ogni tipo di attivita
XP_BASE_REWARDS: Dict[str, int] = {
ActivityType.LESSON_COMPLETED.value: 50,
ActivityType.QUIZ_PASSED.value: 100,
ActivityType.QUIZ_PERFECT.value: 200,
ActivityType.COURSE_COMPLETED.value: 500,
ActivityType.STREAK_BONUS.value: 25,
ActivityType.PEER_REVIEW.value: 75,
ActivityType.HELP_PEER.value: 150,
ActivityType.FIRST_LOGIN_DAY.value: 10,
}
@dataclass
class XPGrant:
student_id: str
activity_type: str
base_xp: int
multiplier: float
final_xp: int
new_total_xp: int
new_level: int
level_up: bool
reason: str
class XPCalculator:
"""
Calcola XP con moltiplicatori basati su streak, difficolta e tempo.
Usa la formula logaritmica per i livelli.
"""
def calculate_level(self, total_xp: int) -> int:
"""
Livello = floor(log2(total_xp / 100)) + 1
Esempi: 100 XP = Lv1, 200 XP = Lv2, 400 XP = Lv3, 800 XP = Lv4...
"""
if total_xp < 100:
return 1
return min(int(math.log2(total_xp / 100)) + 1, 100)
def xp_for_level(self, level: int) -> int:
"""XP necessari per raggiungere il livello."""
if level <= 1:
return 0
return 100 * (2 ** (level - 1))
def calculate_multiplier(
self,
streak_days: int,
difficulty: int, # 1-5
time_of_day_bonus: bool = False,
) -> float:
"""
Calcola il moltiplicatore XP.
- Streak: +5% per ogni giorno consecutivo (max +50%)
- Difficolta: +10% per ogni livello sopra il baseline (2)
- Bonus orario: +20% per studio nelle ore ottimali (8-12, 14-18)
"""
streak_bonus = min(streak_days * 0.05, 0.5) # Max 50%
difficulty_bonus = max(0, (difficulty - 2) * 0.10) # Baseline difficolta = 2
time_bonus = 0.20 if time_of_day_bonus else 0.0
return 1.0 + streak_bonus + difficulty_bonus + time_bonus
def grant_xp(
self,
student_id: str,
activity: ActivityType,
current_xp: int,
streak_days: int,
difficulty: int = 2,
time_of_day_bonus: bool = False,
) -> XPGrant:
base_xp = XP_BASE_REWARDS.get(activity.value, 10)
multiplier = self.calculate_multiplier(streak_days, difficulty, time_of_day_bonus)
final_xp = int(base_xp * multiplier)
new_total = current_xp + final_xp
old_level = self.calculate_level(current_xp)
new_level = self.calculate_level(new_total)
return XPGrant(
student_id=student_id,
activity_type=activity.value,
base_xp=base_xp,
multiplier=multiplier,
final_xp=final_xp,
new_total_xp=new_total,
new_level=new_level,
level_up=new_level > old_level,
reason=f"{activity.value} x{multiplier:.2f} (streak:{streak_days}d, diff:{difficulty}/5)",
)4. System identyfikatorów z idempotencją i konfigurowalnymi regułami
Najczęstszym problemem w systemach identyfikatorów jest podwójne przypisanie: ten sam identyfikator jest przyznawany wielokrotnie za warunki wyścigu lub ponowne nieudane zawody. Rozwiązanie iidempotencja: każda operacja przypisania identyfikatora można wykonać wielokrotnie, zawsze uzyskując ten sam wynik. Używamy A ograniczenie UNIKALNE w PostgreSQL jako zabezpieczenie na poziomie bazy danych.
# gamification/badge_manager.py
import json
from dataclasses import dataclass
from typing import List, Dict, Any, Optional
from datetime import datetime
from enum import Enum
class BadgeTier(Enum):
BRONZE = "bronze"
SILVER = "silver"
GOLD = "gold"
PLATINUM = "platinum"
LEGENDARY = "legendary"
@dataclass
class BadgeDefinition:
badge_id: str
name: str
description: str
tier: BadgeTier
icon_url: str
# Regola come JSON (no hardcoding!)
rule: Dict[str, Any]
# Esempi di regole configurabili
BADGE_CATALOG: List[BadgeDefinition] = [
BadgeDefinition(
badge_id="first_lesson",
name="Primo Passo",
description="Hai completato la tua prima lezione",
tier=BadgeTier.BRONZE,
icon_url="/badges/first-lesson.svg",
rule={"event": "lesson_completed", "count": 1},
),
BadgeDefinition(
badge_id="week_streak",
name="Settimana di Fuoco",
description="7 giorni di studio consecutivi",
tier=BadgeTier.SILVER,
icon_url="/badges/week-streak.svg",
rule={"metric": "streak_days", "gte": 7},
),
BadgeDefinition(
badge_id="perfect_quiz",
name="Perfezionista",
description="Quiz completato con 100% di risposte corrette",
tier=BadgeTier.GOLD,
icon_url="/badges/perfect-quiz.svg",
rule={"event": "quiz_completed", "score": 100},
),
BadgeDefinition(
badge_id="mentor",
name="Mentore della Comunita",
description="Hai aiutato 10 compagni di corso",
tier=BadgeTier.PLATINUM,
icon_url="/badges/mentor.svg",
rule={"metric": "peers_helped", "gte": 10},
),
]
class BadgeManager:
def __init__(self, db, redis_client):
self.db = db
self.redis = redis_client
self._catalog: Dict[str, BadgeDefinition] = {b.badge_id: b for b in BADGE_CATALOG}
async def evaluate_and_award(
self,
student_id: str,
event: str,
student_metrics: Dict[str, Any],
) -> List[str]:
"""
Valuta tutti i badge applicabili e li assegna (idempotente).
Ritorna la lista di badge_id appena assegnati.
"""
newly_awarded = []
for badge_id, badge_def in self._catalog.items():
if await self._is_already_awarded(student_id, badge_id):
continue
if self._evaluate_rule(badge_def.rule, event, student_metrics):
awarded = await self._award_badge_idempotent(student_id, badge_id)
if awarded:
newly_awarded.append(badge_id)
return newly_awarded
async def _award_badge_idempotent(self, student_id: str, badge_id: str) -> bool:
"""
Assegna il badge usando INSERT ... ON CONFLICT DO NOTHING.
Garantisce idempotenza a livello DB.
Ritorna True se il badge e stato effettivamente assegnato ora.
"""
result = await self.db.execute(
"""INSERT INTO student_badges (student_id, badge_id, awarded_at)
VALUES (:sid, :bid, :ts)
ON CONFLICT (student_id, badge_id) DO NOTHING
RETURNING badge_id""",
{"sid": student_id, "bid": badge_id, "ts": datetime.utcnow()},
)
rows = result.fetchall()
if rows:
# Invalida cache badge studente
await self.redis.delete(f"student:badges:{student_id}")
await self.db.commit()
return True
return False # Già esisteva
async def _is_already_awarded(self, student_id: str, badge_id: str) -> bool:
"""Controlla cache Redis prima di interrogare il DB."""
cache_key = f"student:badges:{student_id}"
cached = await self.redis.get(cache_key)
if cached:
badges = json.loads(cached)
return badge_id in badges
# Cache miss: interroga DB
result = await self.db.execute(
"SELECT badge_id FROM student_badges WHERE student_id = :sid",
{"sid": student_id},
)
badges = [row[0] for row in result.fetchall()]
await self.redis.setex(cache_key, 300, json.dumps(badges)) # TTL 5 minuti
return badge_id in badges
def _evaluate_rule(
self,
rule: Dict,
current_event: str,
metrics: Dict[str, Any],
) -> bool:
"""Valuta una regola badge configurabile."""
# Regola basata su evento
if "event" in rule:
if rule["event"] != current_event:
return False
if "count" in rule:
return metrics.get(f"event_count_{rule['event']}", 0) >= rule["count"]
if "score" in rule:
return metrics.get("last_score", 0) >= rule["score"]
return True
# Regola basata su metrica
if "metric" in rule:
metric_val = metrics.get(rule["metric"], 0)
if "gte" in rule:
return metric_val >= rule["gte"]
if "eq" in rule:
return metric_val == rule["eq"]
if "lte" in rule:
return metric_val <= rule["lte"]
return False5. Tablica liderów w czasie rzeczywistym z zestawami posortowanymi Redis
Globalne rankingi to miecz obosieczny: motywują najlepszych ale demotywują tych, którzy znajdują się na dole rankingów. Nowoczesne rozwiązanie to segmentowana tabela wyników: każdy uczeń konkuruje z 50 użytkownikami bliżej jego wyniku, dzięki czemu rywalizacja zawsze ma znaczenie. Narzędziem są zestawy sortowane Redis o złożoności O(log N) dla wszystkich operacji idealny do rankingów w czasie rzeczywistym.
# gamification/leaderboard.py
import json
from typing import List, Dict, Optional
from datetime import datetime, date
import redis.asyncio as redis
@dataclass
class LeaderboardEntry:
rank: int
student_id: str
display_name: str
xp: int
level: int
avatar_url: str
is_current_user: bool = False
class LeaderboardManager:
"""
Leaderboard multi-scope con Redis Sorted Sets.
Scope: global, course-specific, weekly, peer-group.
"""
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
def _key(self, scope: str, period: str = "all-time") -> str:
"""Genera chiave Redis per uno scope e periodo."""
if period == "weekly":
week = date.today().isocalendar()[1]
year = date.today().year
return f"leaderboard:{scope}:{year}:w{week}"
return f"leaderboard:{scope}:{period}"
async def update_score(
self,
student_id: str,
xp_delta: int,
scopes: List[str],
) -> None:
"""
Aggiorna il punteggio in tutti gli scope rilevanti.
Usa pipeline Redis per operazioni atomiche.
"""
async with self.redis.pipeline(transaction=True) as pipe:
for scope in scopes:
for period in ["all-time", "weekly"]:
key = self._key(scope, period)
pipe.zincrby(key, xp_delta, student_id)
# Imposta TTL per leaderboard settimanale
if period == "weekly":
pipe.expire(key, 7 * 24 * 3600)
await pipe.execute()
async def get_leaderboard(
self,
scope: str,
page: int = 1,
page_size: int = 10,
period: str = "all-time",
) -> List[Dict]:
"""Recupera la leaderboard paginata per uno scope."""
key = self._key(scope, period)
start = (page - 1) * page_size
end = start + page_size - 1
# ZREVRANGE: ordine decrescente (punteggio più alto prima)
entries = await self.redis.zrevrange(key, start, end, withscores=True)
result = []
for rank_offset, (student_id_bytes, score) in enumerate(entries):
student_id = student_id_bytes.decode() if isinstance(student_id_bytes, bytes) else student_id_bytes
result.append({
"rank": start + rank_offset + 1,
"student_id": student_id,
"xp": int(score),
})
return result
async def get_peer_leaderboard(
self,
student_id: str,
scope: str,
window: int = 25,
period: str = "all-time",
) -> List[Dict]:
"""
Leaderboard contestualizzata: mostra i 25 utenti sopra e sotto il corrente studente.
"""
key = self._key(scope, period)
# Posizione attuale dello studente (rank 0-indexed dal basso)
rank_from_bottom = await self.redis.zrank(key, student_id)
if rank_from_bottom is None:
return []
total = await self.redis.zcard(key)
rank_from_top = total - rank_from_bottom - 1
# Finestra: window utenti sopra e sotto
start = max(0, rank_from_top - window)
end = min(total - 1, rank_from_top + window)
entries = await self.redis.zrevrange(key, start, end, withscores=True)
result = []
for i, (sid_bytes, score) in enumerate(entries):
sid = sid_bytes.decode() if isinstance(sid_bytes, bytes) else sid_bytes
result.append({
"rank": start + i + 1,
"student_id": sid,
"xp": int(score),
"is_current_user": sid == student_id,
})
return result
async def get_student_rank(
self,
student_id: str,
scope: str,
period: str = "all-time",
) -> Optional[int]:
"""Recupera il rank assoluto di uno studente."""
key = self._key(scope, period)
total = await self.redis.zcard(key)
rank_from_bottom = await self.redis.zrank(key, student_id)
if rank_from_bottom is None:
return None
return total - rank_from_bottom # Rank dal top (1 = primo posto)6. Śledzenie serii i nagrody tymczasowe
Smugi to jedna z najpotężniejszych mechanik zaangażowania w EdTech. Duolingo zbudowało imperium w oparciu o koncepcję „nie przerywaj passy”. Implementacja musi poprawnie obsługiwać strefy czasowe (aplikacja student w Japonii nie może stracić dobrej passy, ponieważ uczył się o 23:59 czasu lokalnego) i „zamraża” (dni łaski na nieoczekiwane zdarzenia).
# gamification/streak_tracker.py
from datetime import datetime, date, timedelta
from typing import Optional, Dict
import zoneinfo
import redis.asyncio as redis
@dataclass
class StreakStatus:
student_id: str
current_streak: int
longest_streak: int
last_activity_date: Optional[date]
freeze_remaining: int # Giorni di grazia disponibili
streak_at_risk: bool # True se non ha ancora fatto attivita oggi
class StreakTracker:
FREEZE_MAX = 2 # Massimo 2 giorni di grazia
def __init__(self, redis_client: redis.Redis, db):
self.redis = redis_client
self.db = db
def _today_for_user(self, timezone: str) -> date:
"""Data corrente nel fuso orario dello studente."""
tz = zoneinfo.ZoneInfo(timezone)
return datetime.now(tz).date()
async def record_activity(
self,
student_id: str,
timezone: str = "UTC",
) -> Dict:
"""
Registra attivita dello studente e aggiorna la streak.
Ritorna le modifiche alla streak.
"""
today = self._today_for_user(timezone)
status = await self.get_status(student_id, timezone)
if not status.last_activity_date:
# Prima attivita in assoluto
new_streak = 1
streak_extended = True
elif status.last_activity_date == today:
# Già registrata oggi: nessun cambio
return {"streak_changed": False, "current_streak": status.current_streak}
elif status.last_activity_date == today - timedelta(days=1):
# Ieri: estendi streak
new_streak = status.current_streak + 1
streak_extended = True
elif (today - status.last_activity_date).days <= 1 + status.freeze_remaining:
# Streak salvata da un freeze
days_missed = (today - status.last_activity_date).days - 1
new_freeze = max(0, status.freeze_remaining - days_missed)
new_streak = status.current_streak + 1
streak_extended = True
await self._update_freeze(student_id, new_freeze)
else:
# Streak rotta
new_streak = 1
streak_extended = False
# Salva aggiornamento
await self._save_streak(student_id, new_streak, status.longest_streak, today)
return {
"streak_changed": True,
"streak_extended": streak_extended,
"current_streak": new_streak,
"longest_streak": max(new_streak, status.longest_streak),
"streak_broken": not streak_extended and status.current_streak > 1,
"previous_streak": status.current_streak if not streak_extended else None,
}
async def get_status(self, student_id: str, timezone: str = "UTC") -> StreakStatus:
"""Recupera lo stato streak corrente dalla cache Redis."""
cache_key = f"streak:{student_id}"
cached = await self.redis.hgetall(cache_key)
if cached:
last_date_str = cached.get(b"last_date", b"").decode()
last_date = date.fromisoformat(last_date_str) if last_date_str else None
today = self._today_for_user(timezone)
return StreakStatus(
student_id=student_id,
current_streak=int(cached.get(b"current", b"0").decode()),
longest_streak=int(cached.get(b"longest", b"0").decode()),
last_activity_date=last_date,
freeze_remaining=int(cached.get(b"freeze", b"0").decode()),
streak_at_risk=last_date != today if last_date else True,
)
# Cache miss: leggi dal DB
row = await self.db.execute(
"""SELECT current_streak, longest_streak, last_activity_date, freeze_remaining
FROM student_streaks WHERE student_id = :sid""",
{"sid": student_id},
)
data = row.fetchone()
if not data:
return StreakStatus(student_id, 0, 0, None, self.FREEZE_MAX, True)
today = self._today_for_user(timezone)
return StreakStatus(
student_id=student_id,
current_streak=data[0],
longest_streak=data[1],
last_activity_date=data[2],
freeze_remaining=data[3],
streak_at_risk=data[2] != today if data[2] else True,
)
async def _save_streak(
self,
student_id: str,
current: int,
longest: int,
last_date: date,
) -> None:
new_longest = max(current, longest)
cache_key = f"streak:{student_id}"
# Aggiorna Redis
await self.redis.hset(cache_key, mapping={
"current": current,
"longest": new_longest,
"last_date": last_date.isoformat(),
})
await self.redis.expire(cache_key, 86400 * 2) # TTL 2 giorni
# Aggiorna DB
await self.db.execute(
"""INSERT INTO student_streaks (student_id, current_streak, longest_streak, last_activity_date)
VALUES (:sid, :cur, :lon, :dt)
ON CONFLICT (student_id) DO UPDATE
SET current_streak = :cur, longest_streak = :lon, last_activity_date = :dt""",
{"sid": student_id, "cur": current, "lon": new_longest, "dt": last_date},
)
await self.db.commit()
async def _update_freeze(self, student_id: str, new_freeze: int) -> None:
await self.redis.hset(f"streak:{student_id}", "freeze", new_freeze)
await self.db.execute(
"UPDATE student_streaks SET freeze_remaining = :fr WHERE student_id = :sid",
{"sid": student_id, "fr": new_freeze},
)
await self.db.commit()Anty-wzorce, których należy unikać
- Unikalna globalna tabela liderów: Zdemotywuj 80% użytkowników, którzy nie znajdują się w pierwszej 10-tce. Używaj segmentów peer-to-peer.
- Odznaki bez znaczenia: Odznaki za każdą błahą czynność nadmuchują system. Odznaki muszą przedstawiać prawdziwe osiągnięcia.
- XP bez ograniczenia prędkości: Boty mogą gromadzić nieskończoną liczbę XP. Wdrażaj ograniczanie szybkości i wykrywanie anomalii.
- Stan niejawny: Zarządzanie postępem za pomocą flag logicznych rozsianych po całym kodzie powoduje błędy, które są trudne do debugowania. Zawsze używaj jawnej maszyny stanów.
- Przypisanie identyfikatora bez idempotencji: Bez ograniczenia UNIKALNE i WSTAW PRZY KONFLIKCIE, ponowne próby mogą spowodować wielokrotne przyznanie tej samej odznaki.
- Seria bez strefy czasowej: Użytkownik w Azji nie powinien stracić passy z powodu problemu z przesunięciem czasu UTC.
Wnioski i dalsze kroki
Zbudowaliśmy podstawy solidnego silnika grywalizacji: jawne maszyny stanowe do progresji, system XP z krzywymi logarytmicznymi, idempotentne odznaki z zasadami konfigurowalne rankingi peer-to-peer w czasie rzeczywistym z Redis i śledzeniem serii z obsługą stref czasowych. Każdy komponent zaprojektowano z myślą o skalowaniu do milionów użytkowników bez pogorszenia wydajności.
Kluczem do sukcesu było podejście konfigurowalne: zasady odznak są one zdefiniowane jako dane JSON, a nie kod zakodowany na stałe. Umożliwia to menedżerom treści aby stworzyć nową mechanikę grywalizacji bez angażowania programistów.
W następnym artykule omówimy Potok Learning Analytics z xAPI i Kafką: jak w jakiś sposób zbierać dane behawioralne od uczniów ujednolicone i przekształcić je w praktyczne spostrzeżenia dla nauczycieli i administratorów.
Seria inżynieryjna EdTech
- Skalowalna architektura LMS: wzorzec wielu najemców
- Algorytmy uczenia się adaptacyjnego: od teorii do produkcji
- Strumieniowe przesyłanie wideo dla edukacji: WebRTC vs HLS vs DASH
- Systemy AI Proctoring: przede wszystkim prywatność dzięki wizji komputerowej
- Spersonalizowany nauczyciel z LLM: RAG dla uziemienia wiedzy
- Silnik grywalizacji: architektura i maszyna stanowa (ten artykuł)
- Learning Analytics: Potok danych z xAPI i Kafką
- Współpraca w czasie rzeczywistym w EdTech: CRDT i WebSocket
- EdTech zorientowany na urządzenia mobilne: architektura zorientowana na tryb offline
- Zarządzanie treścią dla wielu dzierżawców: wersjonowanie i SCORM