feat(recommender): '느림' 판정을 중앙값×1.5 → 난이도별 절대 컷오프로 교체

feat(recommender): '느림' 판정을 중앙값×1.5 → 난이도별 절대 컷오프로 교체

Author: dlxodus02

quiz_recommender/personalize.py

@@ -9,10 +9,11 @@
             상태머신. 난이도 1=하, 2=중, 3=상.
 
 역할분담: 선정(어느 단원을 먼저)은 시간+정오답, 난이도(얼마나 어렵게)는 정오답만.
-'느림' 기준은 학생 본인의 풀이시간 중앙값 × SLOW_FACTOR — 절대초 기준이 아니라
-본인 상대 기준이라 원래 신중하게 푸는 학생이 불이익을 받지 않는다.
-시간 표본이 MIN_TIME_SAMPLES 미만이면(측정 전 데이터) 시간 신호는 자동 비활성 —
-이때는 기존 동작(틀린 단원 우선, 맞힌 문제 전부 제외)과 동일하다.
+'느림' 기준은 문제 난이도별 목표시간(SLOW_CUTOFF_SECONDS) 초과 여부 — 수능 수학의
+하/중/상 시간배분을 근거로 한 절대 기준이다. 난이도가 다르면 정상 풀이시간도 다르므로
+(하는 금방, 상은 킬러라 오래) 하나의 중앙값 잣대로 비교하지 않고 난이도별 컷오프를 각각 적용한다.
+시간이 미측정(None)이거나 난이도가 미상이면 그 문제엔 시간 신호가 붙지 않아 —
+기존 동작(틀린 단원 우선, 맞힌 문제 전부 제외)과 동일해진다.
 
 사다리 상태와 유사 2문제의 난이도 구성:
   BASE  (첫 복습)          → (원 난이도, 원+1)   수준 확인 + 살짝 도전
@@ -28,7 +29,6 @@
 
 원문제 난이도가 없으면(NULL) 사다리를 건너뛰고 약점 재랭킹만 한다.
 """
-import statistics
 from functools import lru_cache
 
 try:
@@ -43,9 +43,13 @@
 MID_DIFFICULTY = 2
 MAX_DIFFICULTY = 3
 
-# 신호③(시간): '느림' = 본인 중앙값 × SLOW_FACTOR 이상. 표본이 적으면 신호 끔.
-SLOW_FACTOR = 1.5
-MIN_TIME_SAMPLES = 3
+# 신호③(시간): '느림' = 문제 난이도별 목표시간(초) 초과. 수능 수학 하/중/상 시간배분 기반.
+# ⚠️ 절대 기준이라 문제가 수능 실전보다 짧으면 거의 안 걸린다 — 실데이터 분포로 캘리브레이션 필요.
+SLOW_CUTOFF_SECONDS = {
+    MIN_DIFFICULTY: 120,   # 하: 2분
+    MID_DIFFICULTY: 240,   # 중: 4분
+    MAX_DIFFICULTY: 600,   # 상: 10분(킬러 — 원래 오래 걸리므로 관대하게)
+}
 
 
 @lru_cache(maxsize=2048)
@@ -86,37 +90,32 @@ def _difficulty_pair(state: str, orig_diff: int) -> tuple[int, int]:
     return (orig_diff, min(orig_diff + 1, MAX_DIFFICULTY))  # BASE(콜드스타트)
 
 
-def _slow_threshold(rounds: list[dict]) -> float | None:
-    """학생 본인 풀이시간 중앙값 × SLOW_FACTOR. 표본 부족(측정 전 데이터)이면 None(신호 끔).
+def _is_slow(qid: int, t: float | None) -> bool:
+    """문제 푸는 데 걸린 시간 t가 그 문제 '난이도별 목표시간'을 넘었으면 '느림'.
 
-    중앙값이 0이면(0초 답이 과반 — 백엔드는 0초를 유효값으로 저장함) threshold=0이 되어
-    측정된 모든 답이 '느림'으로 반전되므로, 이 퇴화 케이스도 None(신호 끔)으로 처리한다."""
-    times = [
-        t for rd in rounds for t in rd.get("times", {}).values() if t is not None
-    ]
-    if len(times) < MIN_TIME_SAMPLES:
-        return None
-    med = statistics.median(times)
-    if med <= 0:
-        return None
-    return med * SLOW_FACTOR
+    난이도가 다르면 정상 풀이시간도 다르므로(하 금방·상 오래) 난이도별 컷오프를 각각 적용한다.
+    시간 미측정(None)·난이도 미상(None)이면 판정 불가 → False(시간 신호 없음)."""
+    if t is None:
+        return False
+    cutoff = SLOW_CUTOFF_SECONDS.get(_difficulty_of(qid))
+    return cutoff is not None and t >= cutoff
 
 
-def _history_signals(rounds: list[dict], threshold: float | None) -> tuple[dict, set]:
+def _history_signals(rounds: list[dict]) -> tuple[dict, set]:
     """이력 → (단원별 우선순위 tier, 완전습득 문제 집합).
 
     tier(작을수록 먼저): 0=틀림+느림, 1=틀림, 2=맞음+느림, 3=무신호.
+    '느림'은 문제 난이도별 목표시간(SLOW_CUTOFF_SECONDS) 초과 여부로 판정(_is_slow).
     mastered: 맞힌 적 있고 한 번도 '느리게 맞은' 적 없는 문제 — 후보에서 제외.
-    threshold=None(시간 미측정)이면 tier 0/2가 나오지 않아 기존 동작과 동일해진다.
+    시간 미측정·난이도 미상이면 slow가 안 나와 tier 0/2가 안 생기고 기존 동작과 동일해진다.
     """
     tiers: dict[int, int] = {}
     correct_fast: set[int] = set()
     correct_slow: set[int] = set()
     for rd in rounds:
         times = rd.get("times", {})
         for qid, ok in rd["answers"]:
-            t = times.get(qid)
-            slow = threshold is not None and t is not None and t >= threshold
+            slow = _is_slow(qid, times.get(qid))
             if ok:
                 (correct_slow if slow else correct_fast).add(qid)
             tier = (0 if slow else 1) if not ok else (2 if slow else 3)
@@ -152,8 +151,7 @@ def personalized_recommend(student_id: int, query_id: int, course_id, k: int) ->
 
     # 2) 필요함(+시간): 오답·풀이시간으로 단원 우선순위와 제외 집합 유도.
     #    '맞음+빠름'만 완전 습득으로 보고 제외 — '맞음+느림'은 후보에 남긴다(3순위).
-    threshold = _slow_threshold(rounds)
-    tiers, mastered = _history_signals(rounds, threshold)
+    tiers, mastered = _history_signals(rounds)
     pool = [c for c in candidates if c not in mastered]
 
     # 3) 수준: 원문제 단원의 라운드들로 사다리 상태 판정 (촉발 라운드 제외)

quiz_recommender/review.py

@@ -5,7 +5,7 @@
 역할 분담 — 종준 FSRS 는 타이밍(언제 뿌릴지)만, 무엇을 복습할지는 여기서 결정한다.
 
 선정 규칙(확정):
-  ① 섹션별 tier + '느림' 기준선 계산 (personalize 재사용)
+  ① 섹션별 tier + 난이도별 '느림' 판정 (personalize 재사용)
   ② 복습 후보 = 이미 푼 문제 중 (틀림+느림 > 틀림 > 맞음+느림). 맞음+빠름(완전습득) 제외
   ③ 급한 순 정렬: 1차 섹션 tier(낮을수록 급함) · 2차 문제 우선순위 · 3차 오래 걸린 순
   ④ 섹션당 원문제 최대 MAX_PER_SECTION, 전체 최대 MAX_TOTAL 까지
@@ -58,8 +58,7 @@ def _select_originals(rounds: list[dict]) -> list[tuple[int, int]]:
     상한: 섹션당 MAX_PER_SECTION, 전체 MAX_TOTAL.
     한 문제가 여러 번 풀렸으면 가장 급한 기록(우선순위 min, 그 안에서 시간 max)으로 대표.
     """
-    threshold = personalize._slow_threshold(rounds)   # '느림' 기준선(없으면 None → 느림 판정 안 함)
-    section_tier, _mastered = personalize._history_signals(rounds, threshold)
+    section_tier, _mastered = personalize._history_signals(rounds)
 
     # 문제별로 가장 급한 기록만 남긴다: {qid: (section_id, priority, time)}
     best: dict[int, tuple[int, int, float]] = {}
@@ -68,7 +67,7 @@ def _select_originals(rounds: list[dict]) -> list[tuple[int, int]]:
         times = rd.get("times", {})
         for qid, ok in rd["answers"]:
             t = times.get(qid)
-            slow = threshold is not None and t is not None and t >= threshold
+            slow = personalize._is_slow(qid, t)   # 난이도별 목표시간 초과 여부
             if ok and not slow:
                 continue  # 맞음+빠름 = 완전 습득 → 원문제 후보 아님
             priority = 0 if (not ok and slow) else 1 if not ok else 2  # 틀림+느림 / 틀림 / 맞음+느림

quiz_recommender/test_recommender.py

@@ -225,6 +225,12 @@ def test_personalize_cold_start_returns_empty():
     301: {"courseId": 1, "sectionId": 11, "difficulty": 2, "instructorId": None},
     302: {"courseId": 1, "sectionId": 22, "difficulty": 2, "instructorId": None},
     303: {"courseId": 1, "sectionId": 33, "difficulty": 2, "instructorId": None},
+    # 이력(history) 문제들 — 난이도 하(1) 고정 → '느림' 컷오프 120초. (tier는 라운드의 section_id로 결정되므로 여기 sectionId는 무관)
+    1: {"courseId": 1, "sectionId": 11, "difficulty": 1, "instructorId": None},
+    2: {"courseId": 1, "sectionId": 11, "difficulty": 1, "instructorId": None},
+    3: {"courseId": 1, "sectionId": 22, "difficulty": 1, "instructorId": None},
+    4: {"courseId": 1, "sectionId": 11, "difficulty": 1, "instructorId": None},
+    5: {"courseId": 1, "sectionId": 11, "difficulty": 1, "instructorId": None},
 }
 
 
@@ -247,9 +253,9 @@ def time_env(monkeypatch):
 def test_wrong_and_slow_section_ranked_first(time_env, monkeypatch):
     """틀림+느림 단원(22)이 틀림-빠름 단원(11)보다 먼저 추천된다."""
     rounds = [
-        # 중앙값: [10,10,90] → 10 → threshold 15
+        # 난이도 하(컷오프 120초): q1@10=빠름, q3@130=느림
         {"section_id": 11, "answers": [(1, False), (2, True)], "times": {1: 10, 2: 10}},
-        {"section_id": 22, "answers": [(3, False)], "times": {3: 90}},   # 틀림+느림 → tier 0
+        {"section_id": 22, "answers": [(3, False)], "times": {3: 130}},   # 틀림+느림 → tier 0
     ]
     monkeypatch.setattr(personalize.db, "get_answer_rounds", lambda sid: rounds)
     picked = personalize.personalized_recommend(7, 100, 1, 2)
@@ -260,9 +266,9 @@ def test_wrong_and_slow_section_ranked_first(time_env, monkeypatch):
 def test_correct_but_slow_not_excluded(time_env, monkeypatch):
     """맞았지만 느린 문제는 완전 습득이 아니므로 후보에서 제외되지 않는다."""
     rounds = [
-        # 301을 맞혔지만 느림(90 ≥ 15) → mastered 아님 → pool에 남음
+        # 301은 난이도 중(컷오프 240초). 맞혔지만 300≥240 → 느림 → mastered 아님 → pool에 남음
         {"section_id": 11, "answers": [(301, True), (4, True), (5, False)],
-         "times": {301: 90, 4: 10, 5: 10}},
+         "times": {301: 300, 4: 10, 5: 10}},
     ]
     monkeypatch.setattr(personalize.db, "get_answer_rounds", lambda sid: rounds)
     picked = personalize.personalized_recommend(7, 100, 1, 3)
@@ -300,22 +306,13 @@ def test_times_key_missing_is_tolerated(time_env, monkeypatch):
     assert picked[0] == 302
 
 
-def test_zero_median_disables_time_signal(time_env, monkeypatch):
-    """중앙값 0(0초 답 과반 — 백엔드는 0초를 유효값으로 저장) → 신호 끔.
-    threshold=0이면 측정된 전부가 '느림'으로 반전되는 퇴화 케이스 방어."""
+def test_zero_time_is_fast_not_slow(time_env, monkeypatch):
+    """0초(찍고 넘긴 답 — 백엔드는 0초를 유효값으로 저장)는 컷오프 미만이라 '빠름'.
+    맞힌 301은 완전 습득으로 기존처럼 제외된다."""
     rounds = [
-        # times [0,0,90] → 중앙값 0 → 신호 꺼짐 → 맞힌 301은 기존처럼 제외돼야 함
         {"section_id": 11, "answers": [(301, True), (4, True), (5, False)],
          "times": {301: 0, 4: 0, 5: 90}},
     ]
     monkeypatch.setattr(personalize.db, "get_answer_rounds", lambda sid: rounds)
-    assert personalize._slow_threshold(rounds) is None
     picked = personalize.personalized_recommend(7, 100, 1, 3)
-    assert 301 not in picked, "신호 꺼짐 → 맞힌 문제는 기존 동작대로 제외"
-
-
-def test_even_sample_uses_true_median():
-    """짝수 표본은 진짜 중앙값(두 가운데 값 평균) 기준: [10,20,30,40] → 25 × 1.5 = 37.5."""
-    rounds = [{"section_id": 1, "answers": [(1, True), (2, True), (3, True), (4, True)],
-               "times": {1: 10, 2: 20, 3: 30, 4: 40}}]
-    assert personalize._slow_threshold(rounds) == 37.5
+    assert 301 not in picked, "0초 맞힘 = 빠름 = 완전 습득 → 제외"

quiz_recommender/test_review.py

@@ -2,8 +2,9 @@
 
 test_recommender.py 와 동일하게, db/vector_store 는 config(env)·qdrant_client 를 끌어오므로
 import 전에 가짜 모듈로 대체한다 → RDS·Qdrant·키 없이 '선정 규칙(계약)'만 순수 검증.
-_select_originals 자체는 personalize 의 순수함수(_slow_threshold/_history_signals)만 쓰므로
-이력(rounds)을 손으로 만들어 정렬·우선순위·상한·제외 규칙을 그대로 못박는다.
+_select_originals 는 personalize._is_slow(난이도별 '느림' 판정)·_history_signals 를 쓴다.
+난이도는 vector_store.retrieve_meta 로 조회하므로 easy_meta 픽스처로 전 문제를 난이도 하(1)로
+고정한다(컷오프 120초). 이력(rounds)을 손으로 만들어 정렬·우선순위·상한·제외 규칙을 못박는다.
 """
 import sys
 import types
@@ -26,44 +27,57 @@ def _round(section_id, answers, times=None):
     return {"section_id": section_id, "answers": answers, "times": times or {}}
 
 
+@pytest.fixture
+def easy_meta(monkeypatch):
+    """선정 테스트의 모든 문제를 난이도 하(1)로 고정 → '느림' = 120초 초과."""
+    personalize._meta_of.cache_clear()
+    monkeypatch.setattr(
+        personalize.vector_store, "retrieve_meta",
+        lambda qid: {"courseId": 1, "sectionId": 0, "difficulty": 1, "instructorId": None},
+        raising=False,
+    )
+    yield
+    personalize._meta_of.cache_clear()
+
+
 # ---------- _select_originals: 선정 규칙 ----------
 
 def test_coldstart_no_history_selects_nothing():
     """이력 없음 → 복습시킬 근거 없음 → 빈 선정."""
     assert review._select_originals([]) == []
 
 
-def test_wrong_slow_ranks_before_wrong_fast_and_masters_excluded():
+def test_wrong_slow_ranks_before_wrong_fast_and_masters_excluded(easy_meta):
     """한 섹션 안: 틀림+느림(우선순위0)이 틀림+빠름(1)보다 앞. 맞음+빠름은 후보 제외."""
-    # times median=1 → 느림 임계값 1.5. q1@10=느림, q2/q3@1=빠름.
-    rounds = [_round(100, [(1, False), (2, False), (3, True)], {1: 10, 2: 1, 3: 1})]
+    # 난이도 하 컷오프 120초. q1@130=느림, q2/q3@10=빠름.
+    rounds = [_round(100, [(1, False), (2, False), (3, True)], {1: 130, 2: 10, 3: 10})]
     result = review._select_originals(rounds)
     assert result == [(1, 100), (2, 100)]        # 느린 오답이 먼저, 그다음 빠른 오답
     assert 3 not in [qid for qid, _ in result]   # 맞음+빠름(완전습득)은 빠짐
 
 
-def test_correct_slow_is_kept_but_lower_priority_than_wrong_section():
+def test_correct_slow_is_kept_but_lower_priority_than_wrong_section(easy_meta):
     """'맞았지만 느림'은 제외되지 않고 남되(우선순위2), 틀린 섹션(tier↓)보다 뒤로 밀린다."""
-    # times [2,2,10] median=2 → 임계값 3. q2@10=느림.
+    # 컷오프 120초. q2@130=느림.
     rounds = [
-        _round(200, [(1, False), (3, False)], {1: 2, 3: 2}),  # 틀림+빠름 → 섹션 tier 1
-        _round(201, [(2, True)], {2: 10}),                    # 맞음+느림 → 섹션 tier 2
+        _round(200, [(1, False), (3, False)], {1: 10, 3: 10}),  # 틀림+빠름 → 섹션 tier 1
+        _round(201, [(2, True)], {2: 130}),                     # 맞음+느림 → 섹션 tier 2
     ]
     result = review._select_originals(rounds)
     assert (2, 201) in result                    # 맞음+느림도 복습 대상으로 살아남음
     assert result[-1] == (2, 201)                # 단, 틀린 섹션 뒤로
     assert all(sec == 200 for _, sec in result[:2])
 
 
-def test_cap_per_section_limits_to_two():
+def test_cap_per_section_limits_to_two(easy_meta):
     """한 섹션에서 후보가 많아도 섹션당 MAX_ORIGINALS_PER_SECTION(2)까지만."""
     rounds = [_round(300, [(1, False), (2, False), (3, False)], {1: 5, 2: 5, 3: 5})]
     result = review._select_originals(rounds)
     assert len(result) == 2
     assert all(sec == 300 for _, sec in result)
 
 
-def test_cap_total_limits_to_ten():
+def test_cap_total_limits_to_ten(easy_meta):
     """전체 후보가 넘쳐도 MAX_ORIGINALS_TOTAL(10)까지만(섹션 6개×2=12 → 10)."""
     rounds = []
     qid = 1
@@ -78,15 +92,15 @@ def test_cap_total_limits_to_ten():
     assert len(result) == review.MAX_ORIGINALS_TOTAL == 10
 
 
-def test_synthetic_slow_correct_gets_selected_when_no_wrong_crowding():
+def test_synthetic_slow_correct_gets_selected_when_no_wrong_crowding(easy_meta):
     """[가상 시나리오] 실데이터(9231)는 틀린 섹션 7개가 top-10을 다 채워 '느린 정답'
     원문제가 선정까지는 못 갔다. 틀림이 자리를 안 뺏는 학생이면 '맞았지만 느림'이
-    실제로 원문제로 뽑히는가? → 뽑힌다. (본인 median 낮게 유지: 대부분 빠른정답)."""
-    # times median=1 → 임계값 1.5. @15만 느림. 빠른정답(@1)은 완전습득 → 제외.
+    실제로 원문제로 뽑히는가? → 뽑힌다."""
+    # 컷오프 120초. @130만 느림. 빠른정답(@10)은 완전습득 → 제외.
     rounds = [
-        _round(700, [(1, True), (3, True)], {1: 15, 3: 1}),   # q1 맞음+느림(target), q3 맞음+빠름(제외)
-        _round(701, [(2, True), (4, True)], {2: 15, 4: 1}),   # q2 맞음+느림(target), q4 맞음+빠름(제외)
-        _round(702, [(5, True), (6, True)], {5: 1, 6: 1}),    # 완전습득 섹션 → 전부 제외
+        _round(700, [(1, True), (3, True)], {1: 130, 3: 10}),   # q1 맞음+느림(target), q3 맞음+빠름(제외)
+        _round(701, [(2, True), (4, True)], {2: 130, 4: 10}),   # q2 맞음+느림(target), q4 맞음+빠름(제외)
+        _round(702, [(5, True), (6, True)], {5: 10, 6: 10}),    # 완전습득 섹션 → 전부 제외
     ]
     result = review._select_originals(rounds)
     assert result == [(1, 700), (2, 701)]                     # 느린 정답 2개가 실제로 선정됨
@@ -109,7 +123,7 @@ def test_recommend_review_attaches_similars_and_course_without_original(monkeypa
         review.recommender, "get_similar_problems",
         lambda sid, pid, k: [pid, 9001, 9002],   # [원문제, 유사1, 유사2]
     )
-    monkeypatch.setattr(review.personalize, "_meta_of", lambda qid: {"courseId": 88})
+    monkeypatch.setattr(review.personalize, "_meta_of", lambda qid: {"courseId": 88, "difficulty": 1})
     result = review.recommend_review(7, k=2)
     assert result == [
         {"problem_id": 11, "section_id": 500, "course_id": 88, "similar": [9001, 9002]}