회계사의 직관을 수치화하다, Python 이상거래/부정 탐지

# 1. 특정 금액 이상 + 계정과목 필터링
condition = (df["amount"] >= 5_000_000) & (df["account"] == "접대비")
high_risk_rule = df[condition]

# 2. 심야(23시~06시) 및 주말 거래 탐지
df["hour"] = pd.to_datetime(df["transaction_date"]).dt.hour
df["weekday"] = pd.to_datetime(df["transaction_date"]).dt.weekday # 5:토, 6:일

night_weekend = df[(df["hour"] < 6) | (df["hour"] >= 23) | (df["weekday"] >= 5)]

# 1. 완전 동일한 거래 찾기 (날짜, 거래처, 금액이 모두 같음)
duplicates = df[df.duplicated(subset=["transaction_date", "customer", "amount"], keep=False)]

# 2. 거래처별 특정 기간(7일) Rolling Sum (단기간 결제 집중 탐지)
df = df.sort_values("transaction_date")
df["rolling_sum_7d"] = df.groupby("customer")["amount"].rolling(7).sum().reset_index(level=0, drop=True)

# 1. 계정과목 내 평소 대비 몇 배나 큰가?
account_mean = df.groupby("account")["amount"].transform("mean")
df["ratio_to_mean"] = df["amount"] / account_mean  # 예: 10배 이상이면 의심

# 2. Z-score (표준편차 이용)
mean, std = df["amount"].mean(), df["amount"].std()
df["z_score"] = (df["amount"] - mean) / std
outlier_z = df[df["z_score"].abs() > 3] # |Z| > 3 이면 통계적으로 드문 값

# 3. IQR (사분위수 범위 이용 - 비정규 분포인 회계데이터에 더 적합)
Q1 = df["amount"].quantile(0.25)
Q3 = df["amount"].quantile(0.75)
IQR = Q3 - Q1
outlier_iqr = df[df["amount"] > (Q3 + 1.5 * IQR)]

# 금액에서 첫 번째 숫자(1~9)만 추출
df["first_digit"] = df["amount"].astype(str).str.replace(r"\D", "", regex=True).str[0]

# 빈도 분포 확인
distribution = df["first_digit"].value_counts(normalize=True).sort_index()
# -> 이 분포가 이론적 수치(1: 30.1%, 2: 17.6%...)와 현저히 다르면 인위적 조작 의심!

from sklearn.ensemble import IsolationForest

# Feature: 거래금액, 요일, 시간대 등
features = df[["amount", "weekday", "hour"]] 

# 전체 데이터의 1% 정도를 이상치로 간주하도록 모델 설정
model = IsolationForest(contamination=0.01, random_state=42)
df["anomaly_ml"] = model.fit_predict(features)  # -1 반환 시 이상치로 판별

df["risk_score"] = 0
df["risk_reason"] = ""

# 조건별 점수 가산
df.loc[df["amount"] >= 10_000_000, ["risk_score", "risk_reason"]] += [30, "고액거래; "]
df.loc[df["ratio_to_mean"] >= 10, ["risk_score", "risk_reason"]] += [20, "평균대비 10배 초과; "]
df.loc[df["anomaly_ml"] == -1, ["risk_score", "risk_reason"]] += [20, "ML이상치 탐지; "]
df.loc[df["hour"] < 6, ["risk_score", "risk_reason"]] += [10, "심야거래; "]

# 위험 등급 판정
def set_risk_level(score):
    if score >= 70: return "HIGH"
    elif score >= 40: return "MEDIUM"
    return "LOW"

df["risk_level"] = df["risk_score"].apply(set_risk_level)