ca3780aa24
包含 CDC consumer、analytics/mastery/warning service、grpc server、repository、ClickHouse DDL 等
257 lines
8.0 KiB
Python
257 lines
8.0 KiB
Python
"""掌握度计算服务（加权滑动平均 + 遗忘曲线）.
|
||
|
||
算法对齐 02-architecture-design.md §9：
|
||
- WEIGHTED_MOVING_AVG：w_i = 0.6^i，归一化（权重从最近到最远递减）
|
||
- FORGETTING_CURVE：基于遗忘曲线的 max 叠加（P5+ 启用），
|
||
half_life = 30 days，越久未练习掌握度越衰减.
|
||
|
||
输入：学生指定知识点的历史成绩序列（按时间倒序）
|
||
输出：mastery_level (0.0-1.0)
|
||
副作用：
|
||
- 写 mastery_snapshot 表（upsert_mastery_snapshot）
|
||
- 发布 mastery.updated 事件（Outbox 豁免）
|
||
|
||
降级策略：
|
||
- ClickHouse 不可达：跳过计算（返回 None）
|
||
- Kafka 不可达：计算继续，事件发布静默失败
|
||
"""
|
||
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
|
||
import structlog
|
||
|
||
from .config import settings
|
||
from .repository import clickhouse_repository, kafka_producer
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
|
||
def _weighted_moving_avg(scores: list[dict[str, Any]]) -> float | None:
|
||
"""加权滑动平均算法.
|
||
|
||
输入：按时间倒序的成绩列表（最近在前），元素含 score 字段
|
||
输出：归一化的 mastery_level (0.0-1.0)，样本不足返回 None.
|
||
|
||
权重：w_i = decay_base ^ i（i 从 0 开始，最近 attempt 权重最大）
|
||
"""
|
||
if not scores:
|
||
return None
|
||
|
||
# 截取最近 N 次（window_size）
|
||
window = scores[: settings.mastery_window_size]
|
||
if len(window) < settings.mastery_min_samples:
|
||
return None
|
||
|
||
decay = settings.mastery_decay_base
|
||
total_weight = 0.0
|
||
weighted_sum = 0.0
|
||
|
||
for i, record in enumerate(window):
|
||
# 成绩归一化到 0.0-1.0（假设原始分 0-100）
|
||
raw_score = record.get("score", 0.0)
|
||
normalized = min(max(raw_score / 100.0, 0.0), 1.0)
|
||
weight = decay**i
|
||
weighted_sum += normalized * weight
|
||
total_weight += weight
|
||
|
||
if total_weight <= 0:
|
||
return None
|
||
|
||
mastery = weighted_sum / total_weight
|
||
return round(max(0.0, min(1.0, mastery)), 4)
|
||
|
||
|
||
def _forgetting_curve_decay(
|
||
mastery: float,
|
||
last_attempt_at: datetime | None,
|
||
now: datetime | None = None,
|
||
) -> float:
|
||
"""遗忘曲线衰减（P5+ 启用，对齐 02 §9）.
|
||
|
||
距离上次练习越久，mastery 越衰减：
|
||
decayed = mastery * exp(-ln(2) * days_since / half_life)
|
||
|
||
half_life = 30 days（settings.mastery_forgetting_half_life_days）.
|
||
"""
|
||
if last_attempt_at is None:
|
||
return mastery
|
||
|
||
now = now or datetime.now(UTC)
|
||
if last_attempt_at.tzinfo is None:
|
||
last_attempt_at = last_attempt_at.replace(tzinfo=UTC)
|
||
|
||
days_since = (now - last_attempt_at).total_seconds() / 86400
|
||
if days_since <= 0:
|
||
return mastery
|
||
|
||
half_life = settings.mastery_forgetting_half_life_days
|
||
if half_life <= 0:
|
||
return mastery
|
||
|
||
import math
|
||
|
||
decay_factor = math.exp(-math.log(2) * days_since / half_life)
|
||
decayed = mastery * decay_factor
|
||
return round(max(0.0, min(1.0, decayed)), 4)
|
||
|
||
|
||
def classify_mastery_label(level: float) -> str:
|
||
"""掌握度三档分类."""
|
||
if level >= 0.8:
|
||
return "mastered"
|
||
if level >= 0.4:
|
||
return "progressing"
|
||
return "weak"
|
||
|
||
|
||
async def calculate_mastery(
|
||
student_id: str,
|
||
knowledge_point_id: str,
|
||
subject_id: str = "",
|
||
) -> dict[str, Any] | None:
|
||
"""计算学生指定知识点的掌握度.
|
||
|
||
流程：
|
||
1. 查询学生该知识点的历史成绩（按时间倒序）
|
||
2. 加权滑动平均 → mastery_level
|
||
3. 遗忘曲线衰减 → 最终 mastery_level
|
||
4. 写 mastery_snapshot 表
|
||
5. 发布 mastery.updated 事件（Outbox 豁免）
|
||
|
||
返回：
|
||
{
|
||
"student_id": ...,
|
||
"knowledge_point_id": ...,
|
||
"subject_id": ...,
|
||
"mastery_level": 0.0-1.0,
|
||
"previous_level": 0.0-1.0 or None,
|
||
"mastery_label": "mastered" / "progressing" / "weak",
|
||
"degraded": bool,
|
||
}
|
||
ClickHouse 不可达时返回降级骨架（degraded=true）.
|
||
"""
|
||
# 1. 查询历史成绩
|
||
scores = await clickhouse_repository.query_student_scores_by_kp(
|
||
student_id=student_id,
|
||
knowledge_point_id=knowledge_point_id,
|
||
limit=settings.mastery_window_size * 2, # 多取一倍容错
|
||
)
|
||
|
||
if scores is None:
|
||
# ClickHouse 不可达降级
|
||
logger.warning(
|
||
"mastery_calc_clickhouse_unavailable_degraded",
|
||
student_id=student_id,
|
||
knowledge_point_id=knowledge_point_id,
|
||
)
|
||
return {
|
||
"student_id": student_id,
|
||
"knowledge_point_id": knowledge_point_id,
|
||
"subject_id": subject_id,
|
||
"mastery_level": 0.0,
|
||
"previous_level": None,
|
||
"mastery_label": "weak",
|
||
"degraded": True,
|
||
"degraded_reason": "clickhouse_unavailable",
|
||
}
|
||
|
||
# 2. 加权滑动平均
|
||
mastery = _weighted_moving_avg(scores)
|
||
if mastery is None:
|
||
# 样本不足，返回默认值
|
||
return {
|
||
"student_id": student_id,
|
||
"knowledge_point_id": knowledge_point_id,
|
||
"subject_id": subject_id,
|
||
"mastery_level": 0.0,
|
||
"previous_level": None,
|
||
"mastery_label": "weak",
|
||
"degraded": False,
|
||
"degraded_reason": "insufficient_samples",
|
||
}
|
||
|
||
# 3. 遗忘曲线衰减（基于最近一次成绩的时间）
|
||
last_attempt_at = None
|
||
if scores:
|
||
last_attempt_at = scores[0].get("timestamp")
|
||
|
||
mastery_final = _forgetting_curve_decay(mastery, last_attempt_at)
|
||
|
||
# 4. 查询上一次 mastery（用于事件对比）
|
||
previous_snapshot = await clickhouse_repository.query_mastery_snapshot(
|
||
student_id=student_id,
|
||
subject_id=subject_id,
|
||
)
|
||
previous_level: float | None = None
|
||
if previous_snapshot is not None:
|
||
for kp in previous_snapshot.get("knowledgePoints", []):
|
||
if kp.get("knowledge_point_id") == knowledge_point_id:
|
||
previous_level = kp.get("mastery_level")
|
||
break
|
||
|
||
# 5. 写 mastery_snapshot 表
|
||
calculated_at = datetime.now(UTC)
|
||
write_ok = await clickhouse_repository.upsert_mastery_snapshot(
|
||
student_id=student_id,
|
||
knowledge_point_id=knowledge_point_id,
|
||
subject_id=subject_id,
|
||
mastery_level=mastery_final,
|
||
calculated_at=calculated_at,
|
||
calculation_method="weighted_moving_avg",
|
||
)
|
||
if not write_ok:
|
||
logger.warning(
|
||
"mastery_snapshot_write_failed_degraded",
|
||
student_id=student_id,
|
||
knowledge_point_id=knowledge_point_id,
|
||
)
|
||
|
||
# 6. 发布 mastery.updated 事件（Outbox 豁免）
|
||
published = await kafka_producer.publish_mastery_updated(
|
||
student_id=student_id,
|
||
knowledge_point_id=knowledge_point_id,
|
||
mastery_level=mastery_final,
|
||
previous_level=previous_level if previous_level is not None else 0.0,
|
||
)
|
||
|
||
result = {
|
||
"student_id": student_id,
|
||
"knowledge_point_id": knowledge_point_id,
|
||
"subject_id": subject_id,
|
||
"mastery_level": mastery_final,
|
||
"previous_level": previous_level,
|
||
"mastery_label": classify_mastery_label(mastery_final),
|
||
"degraded": not write_ok,
|
||
"degraded_reason": "snapshot_write_failed" if not write_ok else "",
|
||
"event_published": published,
|
||
}
|
||
logger.info(
|
||
"mastery_calculated",
|
||
student_id=student_id,
|
||
knowledge_point_id=knowledge_point_id,
|
||
mastery_level=mastery_final,
|
||
previous_level=previous_level,
|
||
label=result["mastery_label"],
|
||
published=published,
|
||
)
|
||
return result
|
||
|
||
|
||
async def batch_calculate_mastery(
|
||
student_id: str,
|
||
knowledge_point_ids: list[str],
|
||
subject_id: str = "",
|
||
) -> list[dict[str, Any]]:
|
||
"""批量计算学生多个知识点的掌握度."""
|
||
results: list[dict[str, Any]] = []
|
||
for kp_id in knowledge_point_ids:
|
||
result = await calculate_mastery(
|
||
student_id=student_id,
|
||
knowledge_point_id=kp_id,
|
||
subject_id=subject_id,
|
||
)
|
||
if result is not None:
|
||
results.append(result)
|
||
return results
|