from __future__ import annotations """ 파생 변수(Index/Score) 산출 근거 trace — 시나리오 무관. L1 batch_builder 의 선언형 step formula 와 실제 입력값을 풀어, 각 최종 파생 변수가 "어떤 입력으로 어떻게 계산됐는지"를 사람이 읽을 수 있는 구조로 반환한다. (프론트 분석 오버레이의 변수 클릭 → 산식 팝오버용) 반환: { 파생변수명: { "value": float, # 최종 값 "clamp": [lo,hi] | None, # 범위 제한(있으면) "kind": "passthrough" | "weighted_sum", "terms": [ { # 기여 항 "ref": str, # 입력 변수명(base 또는 노출 Index명) "ref_value": float, # 그 입력의 현재 값 "weight": float | None, # 가중치(선형 [w,0] 항일 때) "contribution": float, # 이 항이 결과에 더한 값 }, ... ], } } """ from typing import Any from core.engines import config from core.engines.extract import survey_base from core.engines.formula import eval_formula, _load_py def build_feature_trace(scenario_id: str, answers: dict[str, str]) -> dict[str, dict]: """파생 변수의 산출 근거 trace를 구성한다. L1 batch_builder의 선언형 formula와 실제 입력값을 풀어, 각 최종 파생 변수가 어떤 입력으로 어떻게 계산됐는지를 사람이 읽을 수 있는 구조로 반환한다. Args: scenario_id: 시나리오 ID. answers: 질문 ID → 선택 응답 코드 매핑. Returns: 파생변수명 → {"value", "clamp", "kind", "terms"} 매핑. step이 없으면 빈 dict. """ survey = config.get_survey(scenario_id) spec = config.get_batch_builder(scenario_id) steps = spec.get("steps", []) if not steps: return {} # 0) base feature → 선택한 응답 라벨 (숫자값 대신 "22시 이후" 같은 의미 표시용) base_answer: dict[str, str] = {} for q in survey.get("questions", []): code = answers.get(q["id"]) if code is None: continue opt = next((o for o in q.get("options", []) if o.get("code") == code), None) if not opt: continue for fname in (opt.get("features") or {}): base_answer[fname] = opt.get("label") # 1) 모든 step을 중간값 포함해 평가 (run_batch_builder 와 동일하되 intermediate 보존) feats: dict[str, Any] = dict(survey_base(survey, answers)) for k, v in spec.get("defaults", {}).items(): feats.setdefault(k, v) if "pre_hook" in spec: feats.update(_load_py(spec["pre_hook"])(feats)) inter_names: set[str] = set() step_formula: dict[str, Any] = {} label_map: dict[str, str] = {} # 중간/파생 변수명 → 읽기 쉬운 라벨(step.label) for st in steps: val = eval_formula(st["formula"], feats) if "round" in st: val = round(val, st["round"]) feats[st["name"]] = val step_formula[st["name"]] = st["formula"] if st.get("label"): label_map[st["name"]] = st["label"] if st.get("intermediate"): inter_names.add(st["name"]) # 2) 중간값(_FAT 등) → 그것을 그대로 노출하는 최종 Index명 매핑 (passthrough) inter_to_final: dict[str, str] = {} for st in steps: f = st["formula"] if (not st.get("intermediate") and isinstance(f, dict) and f.get("feat") in inter_names and f.get("linear") == [1, 0] and "div" not in f and "mul" not in f): inter_to_final[f["feat"]] = st["name"] def _terms(formula): """formula의 기여 항 리스트를 추출한다. Args: formula: 평가 대상 formula 노드. Returns: 기여 항 리스트. 상수 등 비-dict formula는 빈 리스트, 단일 feat 노드는 그 노드만. """ if not isinstance(formula, dict): return [] # 상수 등 비-dict formula → 기여 항 없음 if "terms" in formula: return formula["terms"] return [formula] # 단일 feat 노드 def _cond_feat(c): """조건(cond)이 참조하는 입력 feat를 찾는다. 단일 조건과 복합 조건(all·any·not)을 모두 탐색한다. Args: c: 조건 노드. Returns: 참조 입력 feat명. 없으면 None. """ if not isinstance(c, dict): return None if c.get("feat") is not None: return c["feat"] for key in ("all", "any"): for sub in (c.get(key) or []): r = _cond_feat(sub) if r: return r if "not" in c: return _cond_feat(c["not"]) return None def _feat_of(t): """항이 참조하는 입력 feat를 찾는다. 평면 {feat}, 중첩 {clamp,value:{feat}}, 조건부 {if/switch}를 모두 인식한다. Args: t: 기여 항 노드. Returns: 참조 입력 feat명. 없으면 None. """ if not isinstance(t, dict): return None if t.get("feat") is not None: return t["feat"] v = t.get("value") if isinstance(v, dict) and v.get("feat") is not None: return v["feat"] if isinstance(t.get("if"), dict): # 조건부(if) 항 → 조건이 보는 입력 return _cond_feat(t["if"]) if isinstance(t.get("switch"), list) and t["switch"]: # switch 항 → 첫 분기 조건의 입력 return _cond_feat(t["switch"][0].get("if", {})) return None def _weight(term: dict): """기여 항의 가중치를 추출한다. 선형 [w, 0] 항의 계수 또는 clamp/value 래퍼 항의 mul을 가중치로 본다. Args: term: 기여 항 노드. Returns: 가중치 값. 가중치로 해석할 수 없으면 None. """ if not isinstance(term, dict): return None lin = term.get("linear") if lin and lin[1] == 0 and "div" not in term and "mul" not in term: return lin[0] if isinstance(term.get("mul"), (int, float)): # clamp/value 래퍼 항은 mul이 사실상 가중치 return term["mul"] return None def _num(v): """수치값을 소수점 2자리로 반올림한다. Args: v: 변환할 값. Returns: 수치(bool 제외)는 반올림한 float, 그 외는 원본 그대로. """ return round(float(v), 2) if isinstance(v, (int, float)) and not isinstance(v, bool) else v trace: dict[str, dict] = {} for st in steps: if st.get("intermediate"): continue name = st["name"] f = st["formula"] # passthrough Index 는 중간값 산식을 인라인해 base 입력까지 노출 eff = f kind = "weighted_sum" if isinstance(f, dict) and f.get("feat") in inter_names and f.get("linear") == [1, 0]: eff = step_formula[f["feat"]] kind = "passthrough" clamp = eff["clamp"] if isinstance(eff, dict) and "clamp" in eff else None # 외부 배수/제수(예: Index의 mul:100) — 각 항 기여에도 동일 적용해야 합이 value와 일치 mul_o = eff.get("mul", 1) if isinstance(eff, dict) else 1 div_o = eff.get("div", 1) if isinstance(eff, dict) else 1 factor = (mul_o or 1) / (div_o or 1) terms = [] for t in _terms(eff): ref = _feat_of(t) try: contribution = round(eval_formula(t, feats) * factor, 2) except Exception: contribution = None if ref is None: # feat 없는 항(상수=기본 점수 / if·switch=조건부) — 합 보존을 위해 포함 is_cond = isinstance(t, dict) and ("if" in t or "switch" in t) terms.append({ "ref": "조건부 항" if is_cond else "기본 점수", "ref_value": None, "ref_answer": None, "weight": None, "contribution": contribution, }) continue terms.append({ "ref": label_map.get(ref) or inter_to_final.get(ref, ref), "ref_value": _num(feats.get(ref, 0.0)), # 문자열(범주형) 값은 그대로 "ref_answer": base_answer.get(ref), # base 입력이면 선택 응답 라벨 "weight": _weight(t), "contribution": contribution, }) trace[name] = { "value": _num(feats[name]), "clamp": clamp, "kind": kind, "terms": terms, } return trace