327 lines
14 KiB
Python
327 lines
14 KiB
Python
import pandas as pd
|
||
import numpy as np
|
||
|
||
# ========== Settings ==========
|
||
|
||
INPUT_CSV = "raw.csv"
|
||
OUTPUT_CSV = "res.csv"
|
||
|
||
# 제외 기준
|
||
CR_THRESHOLD = 0.10
|
||
INTUITION_SD_MIN = 0.5
|
||
PAIRWISE_SAME_ALL = True
|
||
|
||
# 불일치 판정 기준
|
||
MISMATCH_THRESHOLD = 2
|
||
|
||
# 모델 점수 → 리커트 변환 경계값; 점수(0~100) → 1~5점
|
||
SCORE_BREAKS = [28, 42, 58, 72]
|
||
|
||
# 사례 카드 15개
|
||
CASES = [
|
||
{"id": 1, "안정성": 85, "수익_가능성": 82, "자금_효율성": 80, "사업_계획_완성도": 88, "사회적_가치": 84, "type": "baseline"},
|
||
{"id": 2, "안정성": 18, "수익_가능성": 22, "자금_효율성": 15, "사업_계획_완성도": 20, "사회적_가치": 25, "type": "baseline"},
|
||
{"id": 3, "안정성": 68, "수익_가능성": 72, "자금_효율성": 65, "사업_계획_완성도": 70, "사회적_가치": 66, "type": "baseline"},
|
||
{"id": 4, "안정성": 35, "수익_가능성": 38, "자금_효율성": 42, "사업_계획_완성도": 30, "사회적_가치": 35, "type": "baseline"},
|
||
{"id": 5, "안정성": 5, "수익_가능성": 68, "자금_효율성": 72, "사업_계획_완성도": 70, "사회적_가치": 65, "type": "spike"},
|
||
{"id": 6, "안정성": 70, "수익_가능성": 97, "자금_효율성": 68, "사업_계획_완성도": 65, "사회적_가치": 72, "type": "spike"},
|
||
{"id": 7, "안정성": 65, "수익_가능성": 70, "자금_효율성": 4, "사업_계획_완성도": 68, "사회적_가치": 72, "type": "spike"},
|
||
{"id": 8, "안정성": 68, "수익_가능성": 65, "자금_효율성": 70, "사업_계획_완성도": 96, "사회적_가치": 66, "type": "spike"},
|
||
{"id": 9, "안정성": 72, "수익_가능성": 68, "자금_효율성": 65, "사업_계획_완성도": 70, "사회적_가치": 6, "type": "spike"},
|
||
{"id": 10, "안정성": 95, "수익_가능성": 95, "자금_효율성": 15, "사업_계획_완성도": 20, "사회적_가치": 25, "type": "conflict"},
|
||
{"id": 11, "안정성": 20, "수익_가능성": 18, "자금_효율성": 22, "사업_계획_완성도": 25, "사회적_가치": 95, "type": "conflict"},
|
||
{"id": 12, "안정성": 88, "수익_가능성": 85, "자금_효율성": 90, "사업_계획_완성도": 92, "사회적_가치": 12, "type": "conflict"},
|
||
{"id": 13, "안정성": 15, "수익_가능성": 92, "자금_효율성": 18, "사업_계획_완성도": 20, "사회적_가치": 88, "type": "conflict"},
|
||
{"id": 14, "안정성": 90, "수익_가능성": 22, "자금_효율성": 88, "사업_계획_완성도": 85, "사회적_가치": 82, "type": "conflict"},
|
||
{"id": 15, "안정성": 48, "수익_가능성": 52, "자금_효율성": 50, "사업_계획_완성도": 47, "사회적_가치": 53, "type": "conflict"},
|
||
]
|
||
|
||
VARS = ["안정성", "수익_가능성", "자금_효율성", "사업_계획_완성도", "사회적_가치"]
|
||
PAIR_IDX = [(0,1),(0,2),(0,3),(0,4),(1,2),(1,3),(1,4),(2,3),(2,4),(3,4)]
|
||
RI = {1:0.00, 2:0.00, 3:0.58, 4:0.90, 5:1.12, 6:1.24, 7:1.32, 8:1.41}
|
||
|
||
# ══════════ AHP 계산 함수 ══════════
|
||
|
||
def build_matrix(pair_values: list) -> np.ndarray:
|
||
"""쌍대비교 10개 값 → 5×5 비율 행렬"""
|
||
A = np.ones((5, 5))
|
||
for k, (i, j) in enumerate(PAIR_IDX):
|
||
v = pair_values[k]
|
||
if v < 0: # Left more important
|
||
a_ij = float(abs(v) + 1)
|
||
elif v > 0: # Right more important
|
||
a_ij = 1.0 / (float(v) + 1)
|
||
else: # Same
|
||
a_ij = 1.0
|
||
A[i][j] = a_ij
|
||
A[j][i] = 1.0 / a_ij
|
||
return A
|
||
|
||
|
||
# 근사
|
||
def power_method(A: np.ndarray, max_iter: int = 1000, tol: float = 1e-9) -> np.ndarray:
|
||
n = len(A)
|
||
v = np.ones(n) / n
|
||
for _ in range(max_iter):
|
||
Av = A @ v
|
||
s = Av.sum()
|
||
if s == 0:
|
||
break
|
||
v_new = Av / s
|
||
if np.max(np.abs(v_new - v)) < tol:
|
||
v = v_new
|
||
break
|
||
v = v_new
|
||
return v
|
||
|
||
|
||
# CR, CI, λmax 계산
|
||
def calc_cr(A: np.ndarray, w: np.ndarray) -> tuple:
|
||
n = len(A)
|
||
Aw = A @ w
|
||
ratios = np.where(w > 1e-10, Aw / w, 0.0)
|
||
lam_max = float(ratios.mean())
|
||
CI = (lam_max - n) / (n - 1)
|
||
ri = RI.get(n, 1.12)
|
||
CR = CI / ri if ri > 0 else 0.0
|
||
return float(CR), float(CI), float(lam_max)
|
||
|
||
|
||
# Calculate AHP of single person
|
||
def ahp_single(pair_values: list) -> dict:
|
||
try:
|
||
A = build_matrix(pair_values)
|
||
w = power_method(A)
|
||
CR, CI, lam = calc_cr(A, w)
|
||
return {"weights": w.tolist(), "CR": CR, "CI": CI,
|
||
"lambda_max": lam, "error": None}
|
||
except Exception as e:
|
||
return {"weights": [np.nan]*5, "CR": np.nan,
|
||
"CI": np.nan, "lambda_max": np.nan, "error": str(e)}
|
||
|
||
|
||
# ========== Score / Difference calculate ==========
|
||
|
||
# Score = sum of 변인값 * 가중치
|
||
def model_score(case: dict, weights: list) -> float:
|
||
vals = np.array([case[v] for v in VARS])
|
||
return float(np.dot(vals, weights))
|
||
|
||
|
||
# 0~100 -> 1~5 Score
|
||
def score_to_liker(score: float) -> int:
|
||
if score >= SCORE_BREAKS[3]: return 5
|
||
elif score >= SCORE_BREAKS[2]: return 4
|
||
elif score >= SCORE_BREAKS[1]: return 3
|
||
elif score >= SCORE_BREAKS[0]: return 2
|
||
else: return 1
|
||
|
||
|
||
# ========== Preprocessing ==========
|
||
|
||
def check_exclusion(row: pd.Series,
|
||
pair_cols: list,
|
||
intuit_cols: list,
|
||
cr: float) -> str:
|
||
|
||
# CR
|
||
if not np.isnan(cr) and cr > CR_THRESHOLD:
|
||
return "제거_CR초과"
|
||
|
||
# Standard deviation
|
||
intuit_vals = row[intuit_cols].values.astype(float)
|
||
if not np.any(np.isnan(intuit_vals)):
|
||
if np.std(intuit_vals) < INTUITION_SD_MIN:
|
||
return "제거_직관무성의"
|
||
|
||
# All same
|
||
if PAIRWISE_SAME_ALL:
|
||
pair_vals = row[pair_cols].values.astype(float)
|
||
if not np.any(np.isnan(pair_vals)):
|
||
if np.std(pair_vals) == 0:
|
||
return "제거_쌍대무성의"
|
||
|
||
return "유효"
|
||
|
||
|
||
# ========== Main Code ==========
|
||
|
||
def main():
|
||
# ── 1. 데이터 로드 ────────────────────────────────────────
|
||
print(f"파일 읽는 중: {INPUT_CSV}")
|
||
df = pd.read_csv(INPUT_CSV, encoding="utf-8-sig")
|
||
print(f" 총 {len(df)}명 로드")
|
||
|
||
# 열 자동 탐지
|
||
pair_cols = sorted([c for c in df.columns if c.startswith("AHP_")])
|
||
|
||
# 직관판단 열: 다양한 형식 자동 탐지
|
||
# 형식 A: "직관판단_Q01" / 형식 B: "직관_사례01" / 형식 C: "직관_Q01"
|
||
intuit_cols = sorted([c for c in df.columns
|
||
if c.startswith("직관판단_Q")
|
||
or c.startswith("직관_사례")
|
||
or c.startswith("직관_Q")])
|
||
|
||
if len(pair_cols) != 10:
|
||
raise ValueError(f"쌍대비교 열이 10개여야 함 (현재 {len(pair_cols)}개). "
|
||
f"열 이름이 'AHP_01_...' 형식인지 확인.")
|
||
if len(intuit_cols) != 15:
|
||
raise ValueError(f"직관판단 열이 15개여야 함 (현재 {len(intuit_cols)}개). "
|
||
f"열 이름이 '직관판단_Q01' / '직관_사례01' / '직관_Q01' 중 하나여야 함.")
|
||
|
||
print(f" 쌍대비교 열: {pair_cols[0]} ~ {pair_cols[-1]}")
|
||
print(f" 직관판단 열: {intuit_cols[0]} ~ {intuit_cols[-1]}")
|
||
|
||
# ── 2. AHP 계산 (전원) ────────────────────────────────────
|
||
print("\nAHP 계산 중...")
|
||
ahp_results = []
|
||
for _, row in df.iterrows():
|
||
pv = [int(row[c]) if pd.notna(row[c]) else 0 for c in pair_cols]
|
||
res = ahp_single(pv)
|
||
ahp_results.append(res)
|
||
|
||
# ── 3. 전처리 판정 ────────────────────────────────────────
|
||
exclusion_flags = []
|
||
for idx, (_, row) in enumerate(df.iterrows()):
|
||
cr = ahp_results[idx]["CR"]
|
||
flag = check_exclusion(row, pair_cols, intuit_cols, cr)
|
||
exclusion_flags.append(flag)
|
||
|
||
df["전처리결과"] = exclusion_flags
|
||
|
||
print("\n[ 전처리 결과 ]")
|
||
for k, v in df["전처리결과"].value_counts().items():
|
||
print(f" {k}: {v}명")
|
||
|
||
# ── 4. 유효 응답자만 추출 ─────────────────────────────────
|
||
valid_mask = df["전처리결과"] == "유효"
|
||
df_valid = df[valid_mask].copy()
|
||
print(f"\n → 최종 분석 대상: {len(df_valid)}명")
|
||
|
||
# ── 5. 결과 컬럼 초기화 ──────────────────────────────────
|
||
result_rows = []
|
||
|
||
for idx in df_valid.index:
|
||
row = df.loc[idx]
|
||
ahp_res = ahp_results[list(df.index).index(idx)]
|
||
weights = ahp_res["weights"]
|
||
|
||
result = {"피실험자ID": row.get("피실험자ID", idx + 1)}
|
||
|
||
# 가중치
|
||
for i, v in enumerate(VARS):
|
||
result[f"가중치_{v}"] = round(weights[i], 4)
|
||
|
||
result["CR"] = round(ahp_res["CR"], 4)
|
||
result["CI"] = round(ahp_res["CI"], 4)
|
||
result["lambda_max"] = round(ahp_res["lambda_max"], 4)
|
||
|
||
# ── 6. 사례별 모델 점수 / 판단 / 불일치 ──────────────
|
||
mis_total = 0
|
||
mis_baseline = 0
|
||
mis_spike = 0
|
||
mis_conflict = 0
|
||
n_baseline = 0
|
||
n_spike = 0
|
||
n_conflict = 0
|
||
|
||
for k_case, case in enumerate(CASES):
|
||
cid = case["id"]
|
||
|
||
# 직관 판단 — intuit_cols 정렬 순서로 접근
|
||
intuit_col = intuit_cols[k_case]
|
||
i_liker = int(row[intuit_col]) if pd.notna(row[intuit_col]) else np.nan
|
||
|
||
# 모델 점수 & 판단
|
||
m_score = model_score(case, weights)
|
||
m_liker = score_to_liker(m_score)
|
||
|
||
# 불일치 판정
|
||
if not np.isnan(i_liker):
|
||
diff = abs(int(i_liker) - m_liker)
|
||
mismatch = int(diff >= MISMATCH_THRESHOLD)
|
||
else:
|
||
mismatch = np.nan
|
||
|
||
result[f"직관_Q{cid:02d}"] = i_liker
|
||
result[f"모델점수_Q{cid:02d}"] = round(m_score, 1)
|
||
result[f"모델판단_Q{cid:02d}"] = m_liker
|
||
result[f"판단차이_Q{cid:02d}"] = (int(i_liker) - m_liker) if not np.isnan(i_liker) else np.nan
|
||
result[f"불일치_Q{cid:02d}"] = mismatch
|
||
|
||
if not np.isnan(mismatch):
|
||
mis_total += mismatch
|
||
ctype = case["type"]
|
||
if ctype == "baseline":
|
||
mis_baseline += mismatch; n_baseline += 1
|
||
elif ctype == "spike":
|
||
mis_spike += mismatch; n_spike += 1
|
||
else:
|
||
mis_conflict += mismatch; n_conflict += 1
|
||
|
||
# ── 7. 불일치율 ───────────────────────────────────────
|
||
result["불일치_합계"] = mis_total
|
||
result["불일치율_전체(%)"] = round(mis_total / 15 * 100, 1)
|
||
result["불일치율_베이스라인(%)"] = round(mis_baseline / n_baseline * 100, 1) if n_baseline else np.nan
|
||
result["불일치율_단일극값(%)"] = round(mis_spike / n_spike * 100, 1) if n_spike else np.nan
|
||
result["불일치율_변인충돌(%)"] = round(mis_conflict / n_conflict * 100, 1) if n_conflict else np.nan
|
||
|
||
# 사후 설문 (있을 경우 포함)
|
||
for q in ["사후Q1_모델과직관차이", "사후Q2_모델기준이해", "사후Q3_모델합리성"]:
|
||
if q in row.index:
|
||
result[q] = row[q]
|
||
|
||
result_rows.append(result)
|
||
|
||
df_out = pd.DataFrame(result_rows)
|
||
|
||
# ── 8. 저장 ──────────────────────────────────────────────
|
||
df_out.to_csv(OUTPUT_CSV, index=False, encoding="utf-8-sig")
|
||
|
||
# ── 9. 요약 출력 ─────────────────────────────────────────
|
||
print("\n" + "=" * 60)
|
||
print("계산 완료 요약")
|
||
print("=" * 60)
|
||
|
||
print(f"\n[ 표본 ]")
|
||
print(f" 전체 입력: {len(df)}명")
|
||
for k, v in df["전처리결과"].value_counts().items():
|
||
print(f" {k}: {v}명")
|
||
print(f" 최종 분석: {len(df_out)}명")
|
||
|
||
print(f"\n[ AHP 가중치 평균 ]")
|
||
for v in VARS:
|
||
m = df_out[f"가중치_{v}"].mean()
|
||
print(f" {v}: {m:.4f} ({m*100:.1f}%)")
|
||
|
||
print(f"\n[ CR 분포 ]")
|
||
cr_vals = df_out["CR"].dropna()
|
||
print(f" 평균: {cr_vals.mean():.4f} 최소: {cr_vals.min():.4f} 최대: {cr_vals.max():.4f}")
|
||
|
||
print(f"\n[ 불일치율 평균 ]")
|
||
for col in ["불일치율_전체(%)", "불일치율_베이스라인(%)",
|
||
"불일치율_단일극값(%)", "불일치율_변인충돌(%)"]:
|
||
m = df_out[col].mean()
|
||
s = df_out[col].std()
|
||
print(f" {col}: M={m:.1f}% SD={s:.1f}%")
|
||
|
||
print(f"\n[ 사례별 불일치율 ]")
|
||
type_label = {"baseline": "베이스라인", "spike": "단일극값", "conflict": "변인충돌"}
|
||
for case in CASES:
|
||
cid = case["id"]
|
||
col = f"불일치_Q{cid:02d}"
|
||
mis = df_out[col].mean() * 100
|
||
i_m = df_out[f"직관_Q{cid:02d}"].mean()
|
||
m_m = df_out[f"모델판단_Q{cid:02d}"].mean()
|
||
tl = type_label[case["type"]]
|
||
print(f" 사례{cid:02d}({tl}): 불일치={mis:4.0f}% 직관={i_m:.2f} 모델={m_m:.2f}")
|
||
|
||
print(f"\n저장 완료: {OUTPUT_CSV}")
|
||
print(f" {len(df_out)}명 * {len(df_out.columns)}열")
|
||
|
||
return df_out
|
||
|
||
|
||
if __name__ == "__main__":
|
||
df_result = main()
|