콘텐츠로 이동

hossam 패키지

hossam

load_info 

load_info(search=None, local=None)

메타데이터에서 사용 가능한 데이터셋 정보를 로드한다.

Parameters:

Name Type Description Default
search str

이름 필터 문자열. 포함하는 항목만 반환.

 None
local str

로컬 메타데이터 경로. None이면 원격(BASE_URL) 사용.

 None

Returns:

Name Type Description
DataFrame DataFrame

name, chapter, desc, url 컬럼을 갖는 테이블

Examples:

from hossam import *
info = load_info()
list(info.columns) #['name', 'chapter', 'desc', 'url']
Source code in hossam/hs_util.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def load_info(search: str | None = None, local: str | None = None) -> DataFrame:
    """메타데이터에서 사용 가능한 데이터셋 정보를 로드한다.

    Args:
        search (str, optional): 이름 필터 문자열. 포함하는 항목만 반환.
        local (str, optional): 로컬 메타데이터 경로. None이면 원격(BASE_URL) 사용.

    Returns:
        DataFrame: name, chapter, desc, url 컬럼을 갖는 테이블

    Examples:
        ```python
        from hossam import *
        info = load_info()
        list(info.columns) #['name', 'chapter', 'desc', 'url']
        ```
    """
    global BASE_URL

    path = None

    if not local:
        data_path = join(BASE_URL, "metadata.json").replace("\\", "/")

        with requests.Session() as session:
            r = session.get(data_path)

            if r.status_code != 200:
                raise Exception("[%d Error] %s ::: %s" % (r.status_code, r.reason, data_path))

        my_dict = r.json()
    else:
        data_path = join(local, "metadata.json")

        if not exists(data_path):
            raise FileNotFoundError("존재하지 않는 데이터에 대한 요청입니다.")

        with open(data_path, "r", encoding="utf-8") as f:
            my_dict = json.loads(f.read())

    my_data = []
    for key in my_dict:
        if 'index' in my_dict[key]:
            del my_dict[key]['index']

        my_dict[key]['url'] = "%s/%s" % (BASE_URL, my_dict[key]['url'])
        my_dict[key]['name'] = key

        if 'chapter' in my_dict[key]:
            my_dict[key]['chapter'] = ", ".join(my_dict[key]['chapter'])
        else:
            my_dict[key]['chapter'] = '공통'

        my_data.append(my_dict[key])

    my_df = DataFrame(my_data)
    my_df2 = my_df.reindex(columns=['name', 'chapter', 'desc', 'url'])

    if search:
        my_df2 = my_df2[my_df2['name'].str.contains(search.lower())]

    return my_df2

load_data 

load_data(key, local=None)

키로 지정된 데이터셋을 로드한다.

Parameters:

Name Type Description Default
key str

메타데이터에 정의된 데이터 식별자(파일명 또는 별칭)

 required
local str

로컬 메타데이터 경로. None이면 원격(BASE_URL) 사용.

 None

Returns:

Type Description
Optional[DataFrame]

DataFrame | None: 성공 시 데이터프레임, 실패 시 None

Examples:

from hossam import *
df = load_data('AD_SALES')  # 메타데이터에 해당 키가 있어야 함
Source code in hossam/hs_util.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
def _load_data_remote(key: str, local: str | None = None) -> Optional[DataFrame]:
    """키로 지정된 데이터셋을 로드한다.

    Args:
        key (str): 메타데이터에 정의된 데이터 식별자(파일명 또는 별칭)
        local (str, optional): 로컬 메타데이터 경로. None이면 원격(BASE_URL) 사용.

    Returns:
        DataFrame | None: 성공 시 데이터프레임, 실패 시 None

    Examples:
        ```python
        from hossam import *
        df = load_data('AD_SALES')  # 메타데이터에 해당 키가 있어야 함
        ```
    """
    index = None
    try:
        url, desc, index = __get_data_url(key, local=local)
    except Exception as e:
        try:
            print(f"\033[91m{str(e)}\033[0m")
        except Exception:
            print(e)
        return

    #print("\033[94m[data]\033[0m", url.replace("\\", "/"))
    #print("\033[94m[desc]\033[0m", desc)
    print(f"\033[94m{desc}\033[0m")

    df = None

    try:
        df = __get_df(url, index_col=index)
    except Exception as e:
        try:
            print(f"\033[91m{str(e)}\033[0m")
        except Exception:
            print(e)
        return


    return df

visualize_silhouette 

visualize_silhouette(
    estimator,
    data,
    xname=None,
    yname=None,
    title=None,
    palette=None,
    outline=True,
    width=config.width,
    height=config.height,
    linewidth=config.line_width,
    save_path=None,
)

군집분석 결과의 실루엣 플롯과 군집 산점도를 한 화면에 함께 시각화함.

수업에서 사용한 visualize_silhouette 함수와 동일한 기능을 수행함.

Parameters:

Name Type Description Default
estimator KMeans | AgglomerativeClustering

학습된 KMeans 또는 AgglomerativeClustering 군집 모델 객체.

 required
data DataFrame

군집분석에 사용된 입력 데이터 (n_samples, n_features).

 required
xname str

산점도 x축에 사용할 컬럼명. None이면 첫 번째 컬럼 사용.

 None
yname str

산점도 y축에 사용할 컬럼명. None이면 두 번째 컬럼 사용.

 None
title str

플롯 제목. None이면 기본값 사용.

 None
palette str

색상 팔레트.

 None
outline bool

산점도 외곽선 표시 여부.

 True
width int

플롯 가로 크기 (inch 단위).

 width
height int

플롯 세로 크기 (inch 단위).

 height
linewidth float

기준선 등 선 두께.

 line_width
save_path str

저장 경로 지정 시 파일로 저장.

 None

Returns:

Type Description
None

None
Note
  • 실루엣 플롯(왼쪽)과 2차원 군집 산점도(오른쪽)를 동시에 확인 가능
  • 군집 품질과 분포를 한눈에 비교·분석할 때 유용
Source code in hossam/hs_plot.py 
2739
2740
2741
2742
2743
2744
2745
2746
2747
2748
2749
2750
2751
2752
2753
2754
2755
2756
2757
2758
2759
2760
2761
2762
2763
2764
2765
2766
2767
2768
2769
2770
2771
2772
2773
2774
2775
2776
2777
2778
2779
2780
2781
2782
2783
2784
2785
2786
2787
2788
2789
2790
2791
2792
2793
2794
2795
2796
2797
2798
2799
2800
def visualize_silhouette(
    estimator: KMeans | AgglomerativeClustering,
    data: DataFrame,
    xname: str | None = None,
    yname: str | None = None,
    title: str | None = None,
    palette: str | None = None,
    outline: bool = True,
    width: int = config.width,
    height: int = config.height,
    linewidth: float = config.line_width,
    save_path: str | None = None,
) -> None:
    """
    군집분석 결과의 실루엣 플롯과 군집 산점도를 한 화면에 함께 시각화함.

    수업에서 사용한 visualize_silhouette 함수와 동일한 기능을 수행함.

    Args:
        estimator (KMeans | AgglomerativeClustering): 학습된 KMeans 또는 AgglomerativeClustering 군집 모델 객체.
        data (DataFrame): 군집분석에 사용된 입력 데이터 (n_samples, n_features).
        xname (str, optional): 산점도 x축에 사용할 컬럼명. None이면 첫 번째 컬럼 사용.
        yname (str, optional): 산점도 y축에 사용할 컬럼명. None이면 두 번째 컬럼 사용.
        title (str, optional): 플롯 제목. None이면 기본값 사용.
        palette (str, optional): 색상 팔레트.
        outline (bool, optional): 산점도 외곽선 표시 여부.
        width (int, optional): 플롯 가로 크기 (inch 단위).
        height (int, optional): 플롯 세로 크기 (inch 단위).
        linewidth (float, optional): 기준선 등 선 두께.
        save_path (str, optional): 저장 경로 지정 시 파일로 저장.

    Returns:
        None

    Note:
        - 실루엣 플롯(왼쪽)과 2차원 군집 산점도(오른쪽)를 동시에 확인 가능
        - 군집 품질과 분포를 한눈에 비교·분석할 때 유용
    """
    fig, ax = get_default_ax(rows=1, cols=2, width=width, height=height, title=title)

    silhouette_plot(
        estimator=estimator,
        data=data,
        ax=ax[0],  # type: ignore
        linewidth=linewidth,
        width=width,
        height=height
    )

    cluster_plot(
        estimator=estimator,
        data=data,
        xname=xname,
        yname=yname,
        ax=ax[1],  # type: ignore
        outline=outline,
        palette=palette,
        width=width,
        height=height
    )

    finalize_plot(ax)

hs_ttest_ind 

hs_ttest_ind(data=None, x=None, y=None, equal_var=None)

두 독립 집단의 평균 차이를 검정한다 (독립표본 t-검정 또는 Welch's t-test).

수업에서 사용한 hs_ttest_ind() 함수를 확장한 버전이다.

독립표본 t-검정은 두 독립된 집단의 평균이 같은지를 검정한다. 귀무가설(H0): μ1 = μ2 (두 집단의 평균이 같다)

Parameters:

Name Type Description Default
data DataFrame | None

x와 y가 컬럼명인 경우 사용할 데이터프레임. 기본값은 None.

 None
x Series | list | ndarray | str | None

첫 번째 집단의 데이터 또는 data가 주어진 경우 연속형 변수의 컬럼명. 기본값은 None.

 None
y Series | list | ndarray | str | None

두 번째 집단의 데이터 또는 data가 주어진 경우 명목형 변수의 컬럼명. 기본값은 None.

 None
equal_var bool | None

등분산성 가정 여부. - True: 독립표본 t-검정 (등분산 가정) - False: Welch's t-test (등분산 가정하지 않음, 더 강건함) - None: equal_var_test()로 자동 판별 기본값은 None.

 None

Returns:

Name Type Description
DataFrame DataFrame

검정 결과를 담은 데이터프레임. 다음 컬럼 포함: - test (str): 사용된 검정 방법 - alternative (str): 대립가설 방향 - statistic (float): t-통계량 - p-value (float): 유의확률 - H0 (bool): 귀무가설 채택 여부 - H1 (bool): 대립가설 채택 여부 - interpretation (str): 검정 결과 해석

Examples:

from hossam import *
from pandas import Series, DataFrame
import numpy as np

# 리스트로 검정
group1 = [5.1, 4.9, 5.3, 5.0, 4.8]
group2 = [5.5, 5.7, 5.4, 5.6, 5.8]
result = hs_stats.ttest_ind(group1, group2)

# Series로 검정
s1 = Series(np.random.normal(5, 1, 100))
s2 = Series(np.random.normal(5.5, 1, 100))
result = hs_stats.ttest_ind(s1, s2, equal_var=False)
Source code in hossam/hs_stats.py 
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
def ttest_ind(
        data: DataFrame | None = None,
        x : Series | list | np.ndarray | str | None = None,
        y : Series | list | np.ndarray | str | None = None,
        equal_var: bool | None = None
) -> DataFrame:
    """두 독립 집단의 평균 차이를 검정한다 (독립표본 t-검정 또는 Welch's t-test).

    수업에서 사용한 hs_ttest_ind() 함수를 확장한 버전이다.

    독립표본 t-검정은 두 독립된 집단의 평균이 같은지를 검정한다.
    귀무가설(H0): μ1 = μ2 (두 집단의 평균이 같다)

    Args:
        data (DataFrame | None, optional): x와 y가 컬럼명인 경우 사용할 데이터프레임.
            기본값은 None.
        x (Series | list | np.ndarray | str | None, optional): 첫 번째 집단의 데이터 또는
            data가 주어진 경우 연속형 변수의 컬럼명. 기본값은 None.
        y (Series | list | np.ndarray | str | None, optional): 두 번째 집단의 데이터 또는
            data가 주어진 경우 명목형 변수의 컬럼명. 기본값은 None.
        equal_var (bool | None, optional): 등분산성 가정 여부.
            - True: 독립표본 t-검정 (등분산 가정)
            - False: Welch's t-test (등분산 가정하지 않음, 더 강건함)
            - None: equal_var_test()로 자동 판별
            기본값은 None.

    Returns:
        DataFrame: 검정 결과를 담은 데이터프레임. 다음 컬럼 포함:
            - test (str): 사용된 검정 방법
            - alternative (str): 대립가설 방향
            - statistic (float): t-통계량
            - p-value (float): 유의확률
            - H0 (bool): 귀무가설 채택 여부
            - H1 (bool): 대립가설 채택 여부
            - interpretation (str): 검정 결과 해석

    Examples:
        ```python
        from hossam import *
        from pandas import Series, DataFrame
        import numpy as np

        # 리스트로 검정
        group1 = [5.1, 4.9, 5.3, 5.0, 4.8]
        group2 = [5.5, 5.7, 5.4, 5.6, 5.8]
        result = hs_stats.ttest_ind(group1, group2)

        # Series로 검정
        s1 = Series(np.random.normal(5, 1, 100))
        s2 = Series(np.random.normal(5.5, 1, 100))
        result = hs_stats.ttest_ind(s1, s2, equal_var=False)
        ```
    """
    # data가 주어지고 x, y가 컬럼명인 경우 데이터 추출
    if data is not None and isinstance(x, str) and isinstance(y, str):
        df = unmelt(data=data, value_vars=x, id_vars=y)
        x = df[df.columns[0]]
        y = df[df.columns[1]]

    # 데이터를 Series로 변환
    if isinstance(x, Series):
        x_data = x.dropna()
    else:
        x_data = Series(x).dropna()

    if isinstance(y, Series):
        y_data = y.dropna()
    else:
        y_data = Series(y).dropna()

    # 데이터 유효성 검사
    if len(x_data) < 2 or len(y_data) < 2:
        raise ValueError(f"각 집단에 최소 2개 이상의 데이터가 필요합니다. x: {len(x_data)}, y: {len(y_data)}")

    # equal_var가 None이면 자동으로 등분산성 판별
    var_checked = False
    if equal_var is None:
        var_checked = True
        # 두 데이터를 DataFrame으로 구성하여 등분산성 검정
        temp_df = DataFrame({'x': x_data, 'y': y_data})
        var_result = equal_var_test(temp_df)
        normality_method = var_result["normality_method"].iloc[0]
        normality_checked = var_result["normality_checked"].iloc[0]
        equal_var_method = var_result["method"].iloc[0]
        equal_var = var_result["is_equal_var"].iloc[0]

    alternative: list = ["two-sided", "less", "greater"]
    result: list = []
    fmt: str = "μ(x) {0} μ(y)"

    for a in alternative:
        try:
            s, p = scipy_ttest_ind(x_data, y_data, equal_var=equal_var, alternative=a)  # type: ignore
            n = "t-test_ind" if equal_var else "Welch's t-test"

            # 검정 결과 해석
            itp = None

            if a == "two-sided":
                itp = fmt.format("==" if p > 0.05 else "!=")    # type: ignore
            elif a == "less":
                itp = fmt.format(">=" if p > 0.05 else "<")     # type: ignore
            else:
                itp = fmt.format("<=" if p > 0.05 else ">")     # type: ignore

            result.append({
                "test": n,
                "alternative": a,
                "interpretation": itp,
                normality_method: normality_checked,
                equal_var_method: equal_var,
                n: round(s, 3),   # type: ignore
                "p-value": round(p, 4),     # type: ignore
                "H0": p > 0.05,             # type: ignore
                "H1": p <= 0.05,            # type: ignore
            })
        except Exception as e:
            result.append({
                "test": "t-test_ind" if equal_var else "Welch's t-test",
                "alternative": a,
                "interpretation": f"검정 실패: {str(e)}",
                normality_method: normality_checked,
                equal_var_method: equal_var,
                n: np.nan,
                "p-value": np.nan,
                "H0": False,
                "H1": False
            })

    rdf = DataFrame(result)
    rdf.set_index(["test", "alternative"], inplace=True)
    return rdf

hs_outlier_table 

hs_outlier_table(data, *fields, columns=None)

데이터프레임의 사분위수와 이상치 경계값, 왜도를 구한다.

수업에서 사용된 hs_outlier_table() 함수를 개선한 버전

Tukey의 방법을 사용하여 각 숫자형 컬럼에 대한 사분위수(Q1, Q2, Q3)와 이상치 판단을 위한 하한(DOWN)과 상한(UP) 경계값을 계산한다. 함수 호출 전 상자그림을 통해 이상치가 확인된 필드에 대해서만 처리하는 것이 좋다.

Parameters:

Name Type Description Default
data DataFrame

분석 대상 데이터프레임.

 required
*fields str

분석할 컬럼명 목록. 지정하지 않으면 모든 숫자형 컬럼을 처리.

 ()
columns list | None

분석할 컬럼명 목록. 지정하지 않으면 모든 숫자형 컬럼을 처리.

 None

Returns:

Name Type Description
DataFrame

각 필드별 사분위수 및 이상치 경계값을 포함한 데이터프레임. 인덱스는 FIELD(컬럼명)이며, 다음 컬럼을 포함:

  • q1 (float): 제1사분위수 (25th percentile)
  • q2 (float): 제2사분위수 (중앙값, 50th percentile)
  • q3 (float): 제3사분위수 (75th percentile)
  • iqr (float): 사분위 범위 (q3 - q1)
  • up (float): 이상치 상한 경계값 (q3 + 1.5 * iqr)
  • down (float): 이상치 하한 경계값 (q1 - 1.5 * iqr)
  • min (float): 최소값
  • max (float): 최대값
  • skew (float): 왜도
  • outlier_count (int): 이상치 개수
  • outlier_rate (float): 이상치 비율(%)

Examples:

from hossam import *
from pandas import DataFrame

df = DataFrame({'x': [1, 2, 3, 100], 'y': [10, 20, 30, 40]})

# 전체 숫자형 컬럼에 대한 이상치 경계 확인:
result = hs_stats.outlier_table(df)
print(result)

# 특정 컬럼만 분석:
result = hs_stats.outlier_table(df, 'x', 'y')
print(result[['q1', 'q3', 'up', 'down']])
Notes
  • DOWN 미만이거나 UP 초과인 값은 이상치(outlier)로 간주됩니다.
  • 숫자형이 아닌 컬럼은 자동으로 제외됩니다.
  • Tukey의 1.5 * IQR 규칙을 사용합니다 (상자그림의 표준 방법).
Source code in hossam/hs_stats.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
def outlier_table(data: DataFrame, *fields: str, columns: list | None = None):
    """데이터프레임의 사분위수와 이상치 경계값, 왜도를 구한다.

    수업에서 사용된 hs_outlier_table() 함수를 개선한 버전    

    Tukey의 방법을 사용하여 각 숫자형 컬럼에 대한 사분위수(Q1, Q2, Q3)와
    이상치 판단을 위한 하한(DOWN)과 상한(UP) 경계값을 계산한다.
    함수 호출 전 상자그림을 통해 이상치가 확인된 필드에 대해서만 처리하는 것이 좋다.

    Args:
        data (DataFrame): 분석 대상 데이터프레임.
        *fields (str): 분석할 컬럼명 목록. 지정하지 않으면 모든 숫자형 컬럼을 처리.
        columns (list | None): 분석할 컬럼명 목록. 지정하지 않으면 모든 숫자형 컬럼을 처리.

    Returns:
        DataFrame: 각 필드별 사분위수 및 이상치 경계값을 포함한 데이터프레임.
            인덱스는 FIELD(컬럼명)이며, 다음 컬럼을 포함:

            - q1 (float): 제1사분위수 (25th percentile)
            - q2 (float): 제2사분위수 (중앙값, 50th percentile)
            - q3 (float): 제3사분위수 (75th percentile)
            - iqr (float): 사분위 범위 (q3 - q1)
            - up (float): 이상치 상한 경계값 (q3 + 1.5 * iqr)
            - down (float): 이상치 하한 경계값 (q1 - 1.5 * iqr)
            - min (float): 최소값
            - max (float): 최대값
            - skew (float): 왜도
            - outlier_count (int): 이상치 개수
            - outlier_rate (float): 이상치 비율(%)

    Examples:
        ```python
        from hossam import *
        from pandas import DataFrame

        df = DataFrame({'x': [1, 2, 3, 100], 'y': [10, 20, 30, 40]})

        # 전체 숫자형 컬럼에 대한 이상치 경계 확인:
        result = hs_stats.outlier_table(df)
        print(result)

        # 특정 컬럼만 분석:
        result = hs_stats.outlier_table(df, 'x', 'y')
        print(result[['q1', 'q3', 'up', 'down']])
        ```

    Notes:
        - DOWN 미만이거나 UP 초과인 값은 이상치(outlier)로 간주됩니다.
        - 숫자형이 아닌 컬럼은 자동으로 제외됩니다.
        - Tukey의 1.5 * IQR 규칙을 사용합니다 (상자그림의 표준 방법).
    """
    if columns is not None:
        if fields:  # type: ignore
            raise ValueError("fields와 columns 인자는 중복 사용할 수 없습니다.")
        fields = columns # type: ignore

    num_columns = list(data.select_dtypes(include=np.number).columns)

    target_fields: list | None = list(fields) if fields else num_columns # type: ignore

    result = []
    for f in target_fields:
        # 숫자 타입이 아니라면 건너뜀
        if f not in num_columns:
            continue

        # 사분위수
        q1 = data[f].quantile(q=0.25)
        q2 = data[f].quantile(q=0.5)
        q3 = data[f].quantile(q=0.75)
        min_value = data[f].min()
        max_value = data[f].max()

        # 이상치 경계 (Tukey's fences)
        iqr = q3 - q1
        down = q1 - 1.5 * iqr
        up = q3 + 1.5 * iqr

        # 왜도
        skew = data[f].skew()

        # 이상치 개수 및 비율
        outlier_count = ((data[f] < down) | (data[f] > up)).sum()
        outlier_rate = (outlier_count / len(data)) * 100

        # 왜도
        skew = data[f].skew()

        # 이상치 개수 및 비율
        outlier_count = ((data[f] < down) | (data[f] > up)).sum()
        outlier_rate = (outlier_count / len(data)) * 100

        # 분포 특성 판정 (왜도 기준)
        abs_skew = abs(skew)    # type: ignore
        if abs_skew < 0.5:      # type: ignore
            dist = "거의 대칭"
        elif abs_skew < 1.0:    # type: ignore
            if skew > 0:        # type: ignore
                dist = "약한 우측 꼬리"
            else:
                dist = "약한 좌측 꼬리"
        elif abs_skew < 2.0:    # type: ignore
            if skew > 0:        # type: ignore
                dist = "중간 우측 꼬리"
            else:
                dist = "중간 좌측 꼬리"
        else:
            if skew > 0:        # type: ignore
                dist = "극단 우측 꼬리"
            else:
                dist = "극단 좌측 꼬리"

        # 로그변환 필요성 판정
        if abs_skew < 0.5:      # type: ignore
            log_need = "낮음"
        elif abs_skew < 1.0:    # type: ignore
            log_need = "중간"
        else:
            log_need = "높음"

        iq = {
            "field": f,
            "q1": q1,
            "q2": q2,
            "q3": q3,
            "iqr": iqr,
            "up": up,
            "down": down,
            "min": min_value,
            "max": max_value,
            "outlier_count": outlier_count,
            "outlier_rate": outlier_rate,
            "skew": skew,
            "dist": dist,
            "log_need": log_need
        }

        result.append(iq)

    return DataFrame(result).set_index("field")

hs_oneway_anova 

hs_oneway_anova(
    data, dv, between, alpha=0.05, posthoc=False
)

일원분산분석(One-way ANOVA)을 일괄 처리한다.

수업에서 사용된 hs_oneway_anova() 함수를 개선한 버전

정규성 및 등분산성 검정을 자동으로 수행한 후, 그 결과에 따라 적절한 ANOVA 방식을 선택하여 분산분석을 수행한다. ANOVA 결과가 유의하면 자동으로 사후검정을 실시한다.

분석 흐름: 1. 정규성 검정 (각 그룹별로 normaltest 수행) 2. 등분산성 검정 (정규성 만족 시 Bartlett, 불만족 시 Levene) 3. ANOVA 수행 (등분산 만족 시 parametric ANOVA, 불만족 시 Welch's ANOVA) 4. ANOVA p-value ≤ alpha 일 때 사후검정 (등분산 만족 시 Tukey HSD, 불만족 시 Games-Howell)

Parameters:

Name Type Description Default
data DataFrame

분석 대상 데이터프레임. 종속변수와 그룹 변수를 포함해야 함.

 required
dv str

종속변수(Dependent Variable) 컬럼명.

 required
between str

그룹 구분 변수 컬럼명.

 required
alpha float

유의수준. 기본값 0.05.

 0.05
posthoc bool

사후검정 수행 여부. 기본값 False.

 False

Returns:

Name Type Description
tuple DataFrame | tuple[DataFrame, DataFrame]
  • anova_df (DataFrame): ANOVA 또는 Welch 결과 테이블(Source, ddof1, ddof2, F, p-unc, np2 등 포함).
  • posthoc_df (DataFrame|None): 사후검정 결과(Tukey HSD 또는 Games-Howell). ANOVA가 유의할 때만 생성.

Examples:

from hossam import *
from pandas import DataFrame

df = DataFrame({
    'score': [5.1, 4.9, 5.3, 5.0, 4.8, 5.5, 5.2, 5.7, 5.3, 5.1],
    'group': ['A', 'A', 'A', 'A', 'A', 'B', 'B', 'B', 'B', 'B']
})

anova_df, posthoc_df = hs_stats.oneway_anova(df, dv='score', between='group')

# 사후검정결과는 ANOVA가 유의할 때만 생성됨
if posthoc_df is not None:
    print(posthoc_report)
    print(posthoc_df.head())

Raises:

Type Description
ValueError

dv 또는 between 컬럼이 데이터프레임에 없을 경우.
Source code in hossam/hs_stats.py 
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
def oneway_anova(data: DataFrame, dv: str, between: str, alpha: float = 0.05, posthoc: bool = False) -> DataFrame | tuple[DataFrame, DataFrame] :
    """일원분산분석(One-way ANOVA)을 일괄 처리한다.

    수업에서 사용된 hs_oneway_anova() 함수를 개선한 버전    

    정규성 및 등분산성 검정을 자동으로 수행한 후,
    그 결과에 따라 적절한 ANOVA 방식을 선택하여 분산분석을 수행한다.
    ANOVA 결과가 유의하면 자동으로 사후검정을 실시한다.

    분석 흐름:
    1. 정규성 검정 (각 그룹별로 normaltest 수행)
    2. 등분산성 검정 (정규성 만족 시 Bartlett, 불만족 시 Levene)
    3. ANOVA 수행 (등분산 만족 시 parametric ANOVA, 불만족 시 Welch's ANOVA)
    4. ANOVA p-value ≤ alpha 일 때 사후검정 (등분산 만족 시 Tukey HSD, 불만족 시 Games-Howell)

    Args:
        data (DataFrame): 분석 대상 데이터프레임. 종속변수와 그룹 변수를 포함해야 함.
        dv (str): 종속변수(Dependent Variable) 컬럼명.
        between (str): 그룹 구분 변수 컬럼명.
        alpha (float, optional): 유의수준. 기본값 0.05.
        posthoc (bool, optional): 사후검정 수행 여부. 기본값 False.

    Returns:
        tuple:
            - anova_df (DataFrame): ANOVA 또는 Welch 결과 테이블(Source, ddof1, ddof2, F, p-unc, np2 등 포함).
            - posthoc_df (DataFrame|None): 사후검정 결과(Tukey HSD 또는 Games-Howell). ANOVA가 유의할 때만 생성.

    Examples:
        ```python
        from hossam import *
        from pandas import DataFrame

        df = DataFrame({
            'score': [5.1, 4.9, 5.3, 5.0, 4.8, 5.5, 5.2, 5.7, 5.3, 5.1],
            'group': ['A', 'A', 'A', 'A', 'A', 'B', 'B', 'B', 'B', 'B']
        })

        anova_df, posthoc_df = hs_stats.oneway_anova(df, dv='score', between='group')

        # 사후검정결과는 ANOVA가 유의할 때만 생성됨
        if posthoc_df is not None:
            print(posthoc_report)
            print(posthoc_df.head())
        ```

    Raises:
        ValueError: dv 또는 between 컬럼이 데이터프레임에 없을 경우.
    """
    # 컬럼 유효성 검사
    if dv not in data.columns:
        raise ValueError(f"'{dv}' 컬럼이 데이터프레임에 없습니다.")
    if between not in data.columns:
        raise ValueError(f"'{between}' 컬럼이 데이터프레임에 없습니다.")

    df_filtered = data[[dv, between]].dropna()

    # ============================================
    # 1. 정규성 검정 (각 그룹별로 수행)
    # ============================================
    group_names = sorted(df_filtered[between].unique())
    normality_satisfied = True

    for group in group_names:
        group_values = df_filtered[df_filtered[between] == group][dv].dropna()
        if len(group_values) > 0:
            s, p = normaltest(group_values)
            if p <= alpha:
                normality_satisfied = False
                break

    # ============================================
    # 2. 등분산성 검정 (그룹별로 수행)
    # ============================================
    # 각 그룹별로 데이터 분리
    group_data_dict = {}
    for group in group_names:
        group_data_dict[group] = df_filtered[df_filtered[between] == group][dv].dropna().values

    # 등분산 검정 수행
    if len(group_names) > 1:
        if normality_satisfied:
            # 정규성을 만족하면 Bartlett 검정
            s, p = bartlett(*group_data_dict.values())
        else:
            # 정규성을 만족하지 않으면 Levene 검정
            s, p = levene(*group_data_dict.values())
        equal_var_satisfied = p > alpha
    else:
        # 그룹이 1개인 경우 등분산성 검정 불가능
        equal_var_satisfied = True

    # ============================================
    # 3. ANOVA 수행
    # ============================================
    anova_df: DataFrame
    anova_method: str

    if equal_var_satisfied:
        # 등분산을 만족할 때 일반적인 ANOVA 사용
        anova_method = "ANOVA"
        anova_df = anova(data=df_filtered, dv=dv, between=between)
        en = "Bartlett"
    else:
        # 등분산을 만족하지 않을 때 Welch's ANOVA 사용
        anova_method = "Welch"
        anova_df = welch_anova(data=df_filtered, dv=dv, between=between)
        en = "Levene"

    # ANOVA 결과에 메타정보 추가
    anova_df.insert(1, 'normality', normality_satisfied)
    anova_df.insert(2, en, equal_var_satisfied)
    anova_df[anova_method] = anova_df['p-unc'] <= alpha if 'p-unc' in anova_df.columns else False  # type: ignore

    if posthoc == False:
        return anova_df

    # ANOVA 결과가 유의한지 확인
    p_unc = float(anova_df.loc[0, 'p-unc']) # type: ignore
    anova_significant = p_unc <= alpha

    # ANOVA 보고 문장 생성
    # def _safe_get(col: str, default: float = np.nan) -> float:
    #     try:
    #         return float(anova_df.loc[0, col]) if col in anova_df.columns else default  # type: ignore
    #     except Exception:
    #         return default

    # df1 = _safe_get('ddof1')
    # df2 = _safe_get('ddof2')
    # fval = _safe_get('F')
    # eta2 = _safe_get('np2')

    # anova_sig_text = "그룹별 평균이 다를 가능성이 높습니다." if anova_significant else "그룹별 평균 차이에 대한 근거가 부족합니다."
    # assumption_text = f"정규성은 {'대체로 만족' if normality_satisfied else '충족되지 않았고'}, 등분산성은 {'충족되었다' if equal_var_satisfied else '충족되지 않았다'}고 판단됩니다."

    # anova_report = (
    #     f"{between}별로 {dv} 평균을 비교한 {anova_method} 결과: F({df1:.3f}, {df2:.3f}) = {fval:.3f}, p = {p_unc:.4f}. "
    #     f"해석: {anova_sig_text} {assumption_text}"
    # )

    # if not np.isnan(eta2):
    #     anova_report += f" 효과 크기(η²p) ≈ {eta2:.3f}, 값이 클수록 그룹 차이가 뚜렷함을 의미합니다."

    # ============================================
    # 4. 사후검정 (ANOVA 유의할 때만)
    # ============================================
    posthoc_df: DataFrame
    posthoc_method: str
    #posthoc_report = "ANOVA 결과가 유의하지 않아 사후검정을 진행하지 않았습니다."

    if anova_significant:
        if equal_var_satisfied:
            # 등분산을 만족하면 Tukey HSD 사용
            posthoc_method = "Tukey HSD"
            posthoc_df = pairwise_tukey(data=df_filtered, dv=dv, between=between)
        else:
            # 등분산을 만족하지 않으면 Games-Howell 사용
            posthoc_method = "Games-Howell"
            posthoc_df = pairwise_gameshowell(df_filtered, dv=dv, between=between)

        # 사후검정 결과에 메타정보 추가
        # posthoc_df.insert(0, 'normality', normality_satisfied)
        # posthoc_df.insert(1, 'equal_var', equal_var_satisfied)
        posthoc_df.insert(0, 'method', posthoc_method)  # type: ignore

        # p-value 컬럼 탐색
        p_cols = [c for c in ["p-tukey", "pval", "p-adjust", "p_adj", "p-corr", "p", "p-unc", "pvalue", "p_value"] if c in posthoc_df.columns]  # type: ignore
        p_col = p_cols[0] if p_cols else None

        # 유의성 여부 컬럼 추가
        posthoc_df['significant'] = posthoc_df[p_col] <= alpha  if p_col else False # type: ignore

        # if p_col:
        #     sig_pairs_df = posthoc_df[posthoc_df[p_col] <= alpha]
        #     sig_count = len(sig_pairs_df)
        #     total_count = len(posthoc_df)
        #     pair_samples = []
        #     if not sig_pairs_df.empty and {'A', 'B'}.issubset(sig_pairs_df.columns):
        #         pair_samples = [f"{row['A']} vs {row['B']}" for _, row in sig_pairs_df.head(3).iterrows()]

        #     if sig_count > 0:
        #         posthoc_report = (
        #             f"{posthoc_method} 사후검정에서 {sig_count}/{total_count}쌍이 의미 있는 차이를 보였습니다 (alpha={alpha})."
        #         )
        #         if pair_samples:
        #             posthoc_report += " 예: " + ", ".join(pair_samples) + " 등."
        #     else:
        #         posthoc_report = f"{posthoc_method} 사후검정에서 추가로 유의한 쌍은 발견되지 않았습니다."
        # else:
        #     posthoc_report = f"{posthoc_method} 결과는 생성했지만 p-value 정보를 찾지 못해 유의성을 확인할 수 없습니다."

    # ============================================
    # 5. 결과 반환
    # ============================================
    #return anova_df, anova_report, posthoc_df, posthoc_report
    return anova_df, posthoc_df

hs_learning_cv 

hs_learning_cv(
    estimator,
    x,
    y,
    scoring=None,
    cv=5,
    train_sizes=np.linspace(0.1, 1.0, 10),
    n_jobs=-1,
)

학습곡선 기반 과적합 판별 함수. 수업에서 사용된 hs_learning_cv 함수와 동일.

Parameters:

Name Type Description Default
estimator

사이킷런 Estimator (파이프라인 권장)

 required
x

설명변수 (DataFrame 또는 ndarray)

 required
y

목표변수 (Series 또는 ndarray)

 required
scoring

평가 지표 (기본값: neg_root_mean_squared_error)

 None
cv

교차검증 폴드 수 (기본값: 5)

 5
train_sizes

학습곡선 학습 데이터 비율 (기본값: np.linspace(0.1, 1.0, 10))

 linspace(0.1, 1.0, 10)
n_jobs

병렬 처리 개수 (기본값: -1, 모든 CPU 사용)

 -1

Returns:

Name Type Description
DataFrame DataFrame

과적합 판별 결과 표
Source code in hossam/hs_ml.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
def learning_cv(
    estimator,
    x,
    y,
    scoring=None,
    cv=5,
    train_sizes=np.linspace(0.1, 1.0, 10),
    n_jobs=-1,
) -> DataFrame:
    """학습곡선 기반 과적합 판별 함수.
    수업에서 사용된 hs_learning_cv 함수와 동일.

    Args:
        estimator: 사이킷런 Estimator (파이프라인 권장)
        x: 설명변수 (DataFrame 또는 ndarray)
        y: 목표변수 (Series 또는 ndarray)
        scoring: 평가 지표 (기본값: neg_root_mean_squared_error)
        cv: 교차검증 폴드 수 (기본값: 5)
        train_sizes: 학습곡선 학습 데이터 비율 (기본값: np.linspace(0.1, 1.0, 10))
        n_jobs: 병렬 처리 개수 (기본값: -1, 모든 CPU 사용)

    Returns:
        DataFrame: 과적합 판별 결과 표
    """

    # -------------------------------------------------
    # [NEW] 문제 유형 자동 판별
    # -------------------------------------------------
    is_classification = (
        hasattr(estimator, "_estimator_type")
        and estimator._estimator_type == "classifier"
    )

    # -------------------------------------------------
    # [NEW] scoring 자동 설정
    # -------------------------------------------------
    if scoring is None:
        scoring = "roc_auc" if is_classification else "neg_root_mean_squared_error"

    # -------------------------------------------------
    # learning curve 계산
    # -------------------------------------------------
    train_sizes, train_scores, cv_scores = learning_curve(  # type: ignore
        estimator=estimator,
        X=x,
        y=y,
        train_sizes=train_sizes,
        cv=cv,
        scoring=scoring,
        n_jobs=n_jobs,
        shuffle=False,
        random_state=52,
    )

    # -------------------------------------------------
    # 모델명 추출 (Pipeline 대응)
    # -------------------------------------------------
    if hasattr(estimator, "named_steps"):
        classname = estimator.named_steps["model"].__class__.__name__
    else:
        classname = estimator.__class__.__name__

    # =================================================
    # 회귀 전용 처리 (기존 코드)
    # =================================================
    if not is_classification:

        # neg RMSE → RMSE
        train_rmse = -train_scores
        cv_rmse = -cv_scores

        # 평균 / 표준편차
        train_mean = train_rmse.mean(axis=1)
        cv_mean = cv_rmse.mean(axis=1)
        cv_std = cv_rmse.std(axis=1)

        # 마지막 지점 기준 정량 판정
        final_train = train_mean[-1]
        final_cv = cv_mean[-1]
        final_std = cv_std[-1]
        gap_ratio = final_train / final_cv
        var_ratio = final_std / final_cv

        # -----------------
        # 과소적합 기준선 (some_threshold)
        # -----------------
        # 기준모형 RMSE (평균 예측)
        y_mean = y.mean()
        rmse_naive = np.sqrt(np.mean((y - y_mean) ** 2))

        # 분산 기반
        std_y = y.std()

        # 최소 설명력(R²) 기반
        min_r2 = 0.10
        rmse_r2 = np.sqrt((1 - min_r2) * np.var(y))

        # 최종 threshold (가장 관대한 기준)
        # -> 원래 some_threshold는 도메인 지식 수준에서 이 모델은 최소 어느 정도의 성능은 내야 한다는 기준을 설정하는 것
        some_threshold = min(rmse_naive, std_y, rmse_r2)

        # -----------------
        # 판정 로직
        # -----------------
        # 마지막 두 지점 기울기
        train_slope = train_mean[-1] - train_mean[-2]
        cv_slope = cv_mean[-1] - cv_mean[-2]

        if gap_ratio >= 0.95 and final_cv > some_threshold:
            status = "⚠️ 과소적합"
        elif gap_ratio <= 0.8 and train_slope > 0 and cv_slope < 0:
            status = "⚠️ 데이터 추가시 일반화 기대"
        elif gap_ratio <= 0.8:
            status = "⚠️ 과대적합"
        elif gap_ratio <= 0.95 and var_ratio <= 0.10:
            status = "✅ 일반화 양호"
        elif var_ratio > 0.15:
            status = "⚠️ 데이터 부족"
        else:
            status = "⚠️ 판단유보"

        # [NEW] 평가 지표 이름
        metric_name = "RMSE"

    # =================================================
    # 분류 전용 처리
    # =================================================
    else:
        # [NEW] 분류는 그대로 사용 (AUC, Accuracy 등)
        train_metric = train_scores
        cv_metric = cv_scores

        train_mean = train_metric.mean(axis=1)
        cv_mean = cv_metric.mean(axis=1)
        cv_std = cv_metric.std(axis=1)

        final_train = train_mean[-1]
        final_cv = cv_mean[-1]
        final_std = cv_std[-1]

        # [NEW] 분류용 비율 정의 (차이 기반)
        gap_ratio = final_train - final_cv
        var_ratio = final_std

        # -----------------
        # [NEW] 분류 판정 로직 (객관적 기준)
        # -----------------
        if final_train < 0.6 and final_cv < 0.6:
            status = "⚠️ 과소적합"
        elif gap_ratio > 0.1:
            status = "⚠️ 과대적합"
        elif gap_ratio <= 0.05 and var_ratio <= 0.05:
            status = "✅ 일반화 양호"
        elif var_ratio > 0.1:
            status = "⚠️ 데이터 부족"
        else:
            status = "⚠️ 판단유보"

        metric_name = scoring.upper()

    # -----------------
    # 정량 결과 표 (수정된 부분)
    # -----------------
    result_df = DataFrame(
        {
            f"Train {metric_name}": [final_train],
            f"CV {metric_name} 평균": [final_cv],
            f"CV {metric_name} 표준편차": [final_std],
            f"Train/CV 비율": [gap_ratio],
            f"CV 변동성 비율": [var_ratio],
            "판정 결과": [status],
        },
        index=[classname],
    )

    # -----------------
    # 학습곡선 시각화
    # -----------------
    def __callback(ax):
        sb.lineplot(
            x=train_sizes,
            y=cv_mean,
            marker="o",
            linewidth=config.line_width,
            markeredgecolor="#ffffff",
            label=f"CV {metric_name}",
        )

        ax.set_xlabel(
            "학습 데이터 비율", fontsize=config.label_font_size
        )  # type : ignore
        ax.set_ylabel(metric_name, fontsize=config.label_font_size)  # type : ignore

    lineplot(
        df=None,
        xname=train_sizes,
        yname=train_mean,
        marker="o",
        markeredgecolor="#ffffff",
        label=f"Train {metric_name}",
        title=f"{classname} 학습곡선 (Learning Curve)",
        callback=__callback,
    )

    return result_df

hs_get_scores 

hs_get_scores(estimator, x_test, y_test)

회귀 성능 평가 지표 함수 수업에서 사용된 hs_get_scores 함수와 동일.

Parameters:

Name Type Description Default
estimator

학습된 사이킷런 회귀 모델

 required
x_test DataFrame

테스트용 설명변수 데이터 (DataFrame)

 required
y_test DataFrame | ndarray

실제 목표변수 값 (DataFrame 또는 ndarray)

 required

Returns:

Name Type Description
DataFrame DataFrame

회귀 성능 평가 지표 (R2, MAE, MSE, RMSE, MAPE, MPE)
Source code in hossam/hs_ml.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def reg_scores(
    estimator, x_test: DataFrame, y_test: DataFrame | np.ndarray
) -> DataFrame:
    """
    회귀 성능 평가 지표 함수
    수업에서 사용된 hs_get_scores 함수와 동일.

    Args:
        estimator: 학습된 사이킷런 회귀 모델
        x_test: 테스트용 설명변수 데이터 (DataFrame)
        y_test: 실제 목표변수 값 (DataFrame 또는 ndarray)

    Returns:
        DataFrame: 회귀 성능 평가 지표 (R2, MAE, MSE, RMSE, MAPE, MPE)
    """
    if hasattr(estimator, "named_steps"):
        classname = estimator.named_steps["model"].__class__.__name__
    else:
        classname = estimator.__class__.__name__

    y_pred = estimator.predict(x_test)

    score_df = DataFrame(
        {
            "결정계수(R2)": r2_score(y_test, y_pred),
            "평균절대오차(MAE)": mean_absolute_error(y_test, y_pred),
            "평균제곱오차(MSE)": mean_squared_error(y_test, y_pred),
            "평균오차(RMSE)": np.sqrt(mean_squared_error(y_test, y_pred)),
            "평균제곱로그오차(MSLE)": mean_squared_log_error(y_test, y_pred),
            "평균로그오차(RMSLE)": np.sqrt(mean_squared_log_error(y_test, y_pred)),
            "평균 절대 백분오차 비율(MAPE)": mean_absolute_percentage_error(
                y_test, y_pred
            ),
            "평균 비율 오차(MPE)": np.mean((y_test - y_pred) / y_test * 100),
        },
        index=[classname],
    )

    return score_df

hs_cls_bin_scores 

hs_cls_bin_scores(
    estimator,
    x_test,
    y_test,
    pos_label=None,
    plot=True,
    plot_width=640,
)

이진 분류 성능 평가 지표 함수.

Parameters:

Name Type Description Default
estimator

학습된 사이킷런 이진 분류 모델

 required
x_test DataFrame

테스트용 설명변수 데이터 (DataFrame)

 required
y_test DataFrame | Series | ndarray

실제 목표변수 값 (DataFrame, Series 또는 ndarray)

 required
pos_label

양성 클래스 레이블 (기본값: None, sklearn 기본 규칙 따름)

 None
plot bool

ROC 곡선 그래프 출력 여부 (기본값: True)

 True
plot_width int | float

ROC 곡선 플롯 너비 (기본값: 640

 640

Returns: DataFrame: 이진 분류 성능 평가 지표 (정확도, 정밀도, 재현율, 위양성율, 특이성, F1 Score, AUC)

Source code in hossam/hs_ml.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def cls_bin_scores(
    estimator,
    x_test: DataFrame,
    y_test: DataFrame | Series | np.ndarray,
    pos_label=None,
    plot: bool = True,
    plot_width: int | float = 640,
) -> DataFrame:
    """
    이진 분류 성능 평가 지표 함수.

    Args:
        estimator: 학습된 사이킷런 이진 분류 모델
        x_test: 테스트용 설명변수 데이터 (DataFrame)
        y_test: 실제 목표변수 값 (DataFrame, Series 또는 ndarray)
        pos_label: 양성 클래스 레이블 (기본값: None, sklearn 기본 규칙 따름)
        plot: ROC 곡선 그래프 출력 여부 (기본값: True)
        plot_width: ROC 곡선 플롯 너비 (기본값: 640
    Returns:
        DataFrame: 이진 분류 성능 평가 지표 (정확도, 정밀도, 재현율, 위양성율, 특이성, F1 Score, AUC)
    """
    # ---------------------------------
    # 1️⃣ 입력 안정성 검사
    # ---------------------------------
    y_test = np.asarray(y_test).ravel()

    if len(np.unique(y_test)) != 2:
        raise ValueError("이 함수는 이항분류 전용임.")

    if not hasattr(estimator, "predict_proba"):
        raise ValueError("predict_proba 지원 모델만 사용 가능.")

    classes = list(estimator.classes_)

    if pos_label is None:
        pos_label = classes[1]  # sklearn 기본 규칙

    if pos_label not in classes:
        raise ValueError("pos_label이 클래스에 존재하지 않음.")

    pos_index = classes.index(pos_label)

    # 위,양성 확률
    y_pred_proba = estimator.predict_proba(x_test)
    # 1로 분류될 확률
    y_pred_proba_1 = estimator.predict_proba(x_test)[:, pos_index]
    # 예측값
    y_pred = estimator.predict(x_test)

    # 혼동행렬
    neg_label = [c for c in classes if c != pos_label][0]
    cm = confusion_matrix(y_test, y_pred, labels=[neg_label, pos_label])

    if cm.shape != (2, 2):
        raise ValueError("혼동행렬이 2x2가 아님.")

    ((TN, FP), (FN, TP)) = cm

    # 클래스 이름
    if hasattr(estimator, "named_steps"):
        classname = estimator.named_steps["model"].__class__.__name__
    else:
        classname = estimator.__class__.__name__

    # auc score
    auc = roc_auc_score(y_test, y_pred_proba_1)

    score_df = DataFrame(
        {
            "정확도(Accuracy)": [accuracy_score(y_test, y_pred)],
            "정밀도(Precision)": [precision_score(y_test, y_pred, pos_label=pos_label, zero_division=0)],
            "재현율(Recall,tpr)": [recall_score(y_test, y_pred, pos_label=pos_label, zero_division=0)],
            "위양성율(Fallout,fpr)": [FP / (TN + FP) if (TN + FP) > 0 else np.nan],
            "특이성(TNR)": [1 - (FP / (TN + FP)) if (TN + FP) > 0 else np.nan],
            "F1 Score": [f1_score(y_test, y_pred, pos_label=pos_label, zero_division=0)],
            "AUC": [auc],
        },
        index=[classname],
    )

    # -----------------------------
    # ROC 곡선 그리기
    # -----------------------------
    if plot:
        roc_fpr, roc_tpr, thresholds = roc_curve(y_test, y_pred_proba_1, pos_label=pos_label)

        def __callback(ax):
            sb.lineplot(x=[0, 1], y=[0, 1], color="red", linestyle=":", alpha=0.5)
            plt.fill_between(x=roc_fpr, y1=roc_tpr, alpha=0.1)  # type: ignore
            ax.set_xlabel(
                "위양성율 (False Positive Rate)", fontsize=config.label_font_size
            )
            ax.set_ylabel(
                "재현율 (True Positive Rate)", fontsize=config.label_font_size
            )
            ax.set_xlim(0, 1)
            ax.set_ylim(0, 1)

        lineplot(
            df=None,
            xname=roc_fpr,
            yname=roc_tpr,
            width=plot_width,           # type: ignore
            height=plot_width * 0.9,    # type: ignore
            title=f"ROC Curve (AUC={auc:.4f})",
            callback=__callback,
        )

    return score_df

hs_get_score_cv 

hs_get_score_cv(
    estimator,
    x_test,
    y_test,
    x_origin,
    y_origin,
    scoring="neg_root_mean_squared_error",
    cv=5,
    train_sizes=np.linspace(0.1, 1.0, 10),
    n_jobs=-1,
)

회귀 성능 평가 지표 함수. 수업에서 사용된 hs_get_score_cv 함수와 동일.

Parameters:

Name Type Description Default
estimator

학습된 사이킷런 회귀 모델

 required
x_test DataFrame

테스트용 설명변수 데이터 (DataFrame)

 required
y_test DataFrame | ndarray

실제 목표변수 값 (DataFrame 또는 ndarray)

 required
x_origin DataFrame

학습곡선용 전체 설명변수 데이터 (DataFrame, learning_curve=True일 때 필요)

 required
y_origin DataFrame | ndarray

학습곡선용 전체 목표변수 값 (DataFrame 또는 ndarray, learning_curve=True일 때 필요)

 required
scoring

학습곡선 평가 지표 (기본값: neg_root_mean_squared_error)

 'neg_root_mean_squared_error'
cv

학습곡선 교차검증 폴드 수 (기본값: 5)

 5
train_sizes

학습곡선 학습 데이터 비율 (기본값: np.linspace(0.1, 1.0, 10))

 linspace(0.1, 1.0, 10)
n_jobs

학습곡선 병렬 처리 개수 (기본값: -1, 모든 CPU 사용)

 -1

Returns:

Name Type Description
DataFrame DataFrame

회귀 성능 평가 지표 + 과적합 판정 여부
Source code in hossam/hs_ml.py 
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
def score_cv(
    estimator,
    x_test: DataFrame,
    y_test: DataFrame | np.ndarray,
    x_origin: DataFrame,
    y_origin: DataFrame | np.ndarray,
    scoring="neg_root_mean_squared_error",
    cv=5,
    train_sizes=np.linspace(0.1, 1.0, 10),
    n_jobs=-1,
) -> DataFrame:
    """
    회귀 성능 평가 지표 함수.
    수업에서 사용된 hs_get_score_cv 함수와 동일.

    Args:
        estimator: 학습된 사이킷런 회귀 모델
        x_test: 테스트용 설명변수 데이터 (DataFrame)
        y_test: 실제 목표변수 값 (DataFrame 또는 ndarray)
        x_origin: 학습곡선용 전체 설명변수 데이터 (DataFrame, learning_curve=True일 때 필요)
        y_origin: 학습곡선용 전체 목표변수 값 (DataFrame 또는 ndarray, learning_curve=True일 때 필요)
        scoring: 학습곡선 평가 지표 (기본값: neg_root_mean_squared_error)
        cv: 학습곡선 교차검증 폴드 수 (기본값: 5)
        train_sizes: 학습곡선 학습 데이터 비율 (기본값: np.linspace(0.1, 1.0, 10))
        n_jobs: 학습곡선 병렬 처리 개수 (기본값: -1, 모든 CPU 사용)

    Returns:
        DataFrame: 회귀 성능 평가 지표 + 과적합 판정 여부
    """

    if type(estimator) != list:
        estimator = [estimator]

    result_df = DataFrame()

    for est in estimator:
        score_df = reg_scores(est, x_test, y_test)
        cv_df = learning_cv(
            est,
            x_origin,
            y_origin,
            scoring=scoring,
            cv=cv,
            train_sizes=train_sizes,
            n_jobs=n_jobs,
        )

        result_df = concat(
            [result_df, merge(score_df, cv_df, left_index=True, right_index=True)]
        )

    return result_df

hs_feature_importance 

hs_feature_importance(
    model,
    x_train,
    y_train,
    threshold=0.9,
    plot=True,
    width=config.width,
)

특징 중요도 분석 함수. 수업에서 사용된 hs_feature_importance 함수와 동일.

Parameters:

Name Type Description Default
model

학습된 사이킷런 회귀 모델

 required
x_train DataFrame

학습용 설명변수 데이터 (DataFrame)

 required
y_train DataFrame | ndarray | Series

학습용 목표변수 값 (DataFrame, Series 또는 ndarray)

 required
threshold float

누적 중요도 기준 (기본값: 0.9)

 0.9
plot bool

중요도 시각화 여부 (기본값: True)

 True
width int

플롯 너비 (기본값: config.width)

 width

Returns:

Name Type Description
DataFrame DataFrame

특징 중요도 결과 표
Source code in hossam/hs_ml.py 
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
def feature_importance(
    model,
    x_train: DataFrame,
    y_train: DataFrame | np.ndarray | Series,
    threshold: float = 0.9,
    plot: bool = True,
    width: int = config.width,
) -> DataFrame:
    """
    특징 중요도 분석 함수.
    수업에서 사용된 hs_feature_importance 함수와 동일.

    Args:
        model: 학습된 사이킷런 회귀 모델
        x_train: 학습용 설명변수 데이터 (DataFrame)
        y_train: 학습용 목표변수 값 (DataFrame, Series 또는 ndarray)
        threshold: 누적 중요도 기준 (기본값: 0.9)
        plot: 중요도 시각화 여부 (기본값: True)
        width: 플롯 너비 (기본값: config.width)

    Returns:
        DataFrame: 특징 중요도 결과 표
    """
    if isinstance(model, XGBRegressor):  # type: ignore
        booster = model.get_booster()  # type: ignore
        imp = booster.get_score(importance_type="gain")
        imp_sr = Series(imp)
        imp_df = DataFrame(imp_sr, columns=["importance"])
    elif isinstance(model, LGBMRegressor):
        booster = model.booster_
        imp = booster.feature_importance(importance_type="gain")
        imp_df = DataFrame({"importance": imp}, index=model.feature_name_)
    elif isinstance(model, CatBoostRegressor):
        imp = model.get_feature_importance(type="FeatureImportance")
        imp_df = DataFrame({"importance": imp}, index=model.feature_names_)
    else:
        imp_df = permutation_importance(
            estimator=model,
            X=x_train,
            y=y_train,
            scoring="r2",
            n_repeats=30,
            random_state=42,
            n_jobs=-1,
        )

        # 결과 정리
        imp_df = DataFrame({"importance": imp_df.importances_mean}, index=x_train.columns)  # type: ignore

    # ----------------------------------
    # 중요도 비율 + 누적 중요도 계산
    # ----------------------------------
    imp_df["ratio"] = imp_df["importance"] / imp_df["importance"].sum()
    imp_df.sort_values("ratio", ascending=False, inplace=True)
    imp_df["cumsum"] = imp_df["ratio"].cumsum()

    # ----------------------------------
    # 시각화
    # ----------------------------------
    if plot:
        df = imp_df.sort_values(by="ratio", ascending=False)

        # 90% 처음 도달하는 인덱스 (0-based)
        cut_idx = np.argmax(imp_df["cumsum"].values >= threshold)  # type: ignore

        # x축은 rank 기준이므로 +1
        cut_rank = (int(cut_idx) + 1) - 0.5

        def __callback(ax):
            # 값 라벨 추가
            for i, v in enumerate(imp_df["importance"]):
                ax.text(
                    v + 0.005,  # 막대 끝에서 약간 오른쪽
                    i,  # y 위치
                    # 표시 형식
                    f"{v:.1f} ({imp_df.iloc[i]['cumsum']*100:.1f}%)",
                    va="center",
                    fontsize=config.text_font_size,
                )

            ax.set_xlabel("importance(cumsum)", fontsize=config.label_font_size)
            ax.set_xlim(0, imp_df["importance"].max() * 1.2)
            ax.set_ylabel(None)

            # 90% 도달 지점 수직선 (핵심)
            plt.axhline(y=cut_rank, linestyle=":", color="red", alpha=0.8)

        barplot(
            df,
            xname="importance",
            yname=df.index,
            width=width,
            height=len(df) * 60,
            title=f"Feature Importance",  # type: ignore
            callback=__callback,
        )

    return imp_df

hs_shap_analysis 

hs_shap_analysis(model, x, plot=True, width=config.width)

SHAP 값 기반 특징 중요도 분석 함수. 수업에서 사용된 hs_shap_analysis 함수와 동일.

Parameters:

Name Type Description Default
model

학습된 사이킷런 회귀 모델

 required
x DataFrame

설명변수 데이터 (DataFrame)

 required
plot bool

SHAP 요약 플롯 시각화 여부 (기본값: True

 True
width int

플롯 너비 (기본값: config.width)

 width

Returns:

Name Type Description
tuple tuple[DataFrame, ndarray]

특징 중요도 요약 DataFrame 및 SHAP 값 배열
Source code in hossam/hs_ml.py 
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
def shap_analysis(
    model,
    x: DataFrame,
    plot: bool = True,
    width: int = config.width,
) -> tuple[DataFrame, np.ndarray]:
    """
    SHAP 값 기반 특징 중요도 분석 함수.
    수업에서 사용된 hs_shap_analysis 함수와 동일.

    Args:
        model: 학습된 사이킷런 회귀 모델
        x: 설명변수 데이터 (DataFrame)
        plot: SHAP 요약 플롯 시각화 여부 (기본값: True
        width: 플롯 너비 (기본값: config.width)

    Returns:
        tuple: 특징 중요도 요약 DataFrame 및 SHAP 값 배열
    """
    # =================================================
    # SHAP 유형 판별
    # =================================================
    # Pipeline 라인 객체와 그렇지 않은 경우 모두 처리
    if isinstance(model, Pipeline):
        estimator = model.named_steps["model"]
        is_pipeline = True
    else:
        estimator = model
        is_pipeline = False

    # 선형 계열인지 아닌지 판별
    is_linear_model = isinstance(
        estimator,
        (
            LinearRegression,
            Ridge,
            Lasso,
            ElasticNet,
            SGDRegressor,
            LogisticRegression,
            SGDClassifier,
        ),
    )

    # 전달된 독립변수 복사
    x_df = x.copy()
    columns = x.columns.tolist()
    indexs = x.index.tolist()

    # 1. SHAP Explainer
    if is_linear_model:
        # pipeline인 경우 파이프라인의 마지막 step을 제외하고 순환하면서 데이터 변환 적용
        if is_pipeline:
            for name, step in list(model.named_steps.items()):
                if name == "model":
                    continue

                x_df = step.transform(x_df)  # type: ignore

        masker = shap.maskers.Independent(x_df)
        explainer = shap.LinearExplainer(estimator, masker=masker)
    else:
        explainer = shap.TreeExplainer(estimator)

    # 2. SHAP 값 계산: shape = [n_samples, n_features]
    shap_values = explainer.shap_values(x_df)

    # 3. DataFrame 변환
    shap_df = DataFrame(
        shap_values,
        columns=columns,
        index=indexs,
    )

    # 4. 요약 통계
    summary_df = DataFrame(
        {
            "feature": shap_df.columns,
            "mean_abs_shap": shap_df.abs().mean().values,
            "mean_shap": shap_df.mean().values,
            "std_shap": shap_df.std().values,
        }
    )

    # 5. 영향 방향 (보수적 표현)
    summary_df["direction"] = np.where(
        summary_df["mean_shap"] > 0,
        "양(+) 경향",
        np.where(summary_df["mean_shap"] < 0, "음(-) 경향", "혼합/미약"),
    )

    # 6. 변동성 지표
    summary_df["cv"] = summary_df["std_shap"] / (summary_df["mean_abs_shap"] + 1e-9)

    summary_df["variability"] = np.where(
        summary_df["cv"] < 1,
        "stable",  # 변동성 낮음 - 평균 대비 일관적 영향 의미
        "variable",  # 변동성 큼 - 상황 의존적 영향 의미
    )

    # 7. 중요도 기준 정렬
    summary_df = summary_df.sort_values("mean_abs_shap", ascending=False).reset_index(
        drop=True
    )

    # 8. 중요 변수 표시 (누적 80%)
    total_importance = summary_df["mean_abs_shap"].sum()
    summary_df["importance_ratio"] = summary_df["mean_abs_shap"] / total_importance
    summary_df["importance_cumsum"] = summary_df["importance_ratio"].cumsum()

    summary_df["is_important"] = np.where(
        summary_df["importance_cumsum"] <= 0.80,
        "core",  # 누적 80% 내 중요 변수 - 모델 핵심 결정 요인 의미 명확
        "secondary",  # 누적 80% 초과 변수 - 보조적/상황적 영향 요인 의미
    )

    # 9. 시각화
    if plot:
        shap.summary_plot(shap_values, x, show=False)

        fig = plt.gcf()
        fig.set_size_inches(width / 100, len(summary_df) * 60 / 100)
        fig.set_dpi(config.dpi)
        ax = fig.get_axes()[0]

        plt.xlabel("SHAP value", fontsize=config.label_font_size)
        plt.xticks(fontsize=config.font_size)
        plt.yticks(fontsize=config.font_size)

        finalize_plot(ax, outparams=True, title="SHAP Summary Plot")

    return summary_df, shap_values

hs_shap_dependence_analysis 

hs_shap_dependence_analysis(
    summary_df,
    shap_values,
    x,
    include_secondary=False,
    width=config.width,
    height=config.height,
)

SHAP 값 기반 특징 의존성 분석 함수 수업에서 사용된 hs_shap_dependence_analysis 함수와 동일.

Parameters:

Name Type Description Default
summary_df DataFrame

shap_analysis 함수의 요약 DataFrame

 required
shap_values ndarray

shap_analysis 함수의 SHAP 값 배열

 required
x DataFrame

설명변수 데이터 (DataFrame)

 required
include_secondary bool

상호작용 변수에 보조 변수 포함 여부 (기본값: False)

 False
width int

플롯 너비 (기본값: config.width)

 width
height int

플롯 높이 (기본값: config.height)

 height

Returns:

Name Type Description
list

생성된 특징 상호작용 쌍 목록
Source code in hossam/hs_ml.py 
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
def shap_dependence_analysis(
    summary_df: DataFrame,
    shap_values: np.ndarray,
    x: DataFrame,
    include_secondary: bool = False,
    width: int = config.width,
    height: int = config.height,
):
    """
    SHAP 값 기반 특징 의존성 분석 함수
    수업에서 사용된 hs_shap_dependence_analysis 함수와 동일.

    Args:
        summary_df: shap_analysis 함수의 요약 DataFrame
        shap_values: shap_analysis 함수의 SHAP 값 배열
        x: 설명변수 데이터 (DataFrame)
        include_secondary: 상호작용 변수에 보조 변수 포함 여부 (기본값: False)
        width: 플롯 너비 (기본값: config.width)
        height: 플롯 높이 (기본값: config.height)

    Returns:
        list: 생성된 특징 상호작용 쌍 목록
    """

    # 1. 주 대상 변수 (Core + Variable)
    main_features = summary_df[
        (summary_df["is_important"] == "core")
        & (summary_df["variability"] == "variable")
    ]["feature"].tolist()

    # 2. 상호작용 후보 변수
    interaction_features = summary_df[summary_df["is_important"] == "core"][
        "feature"
    ].tolist()

    if include_secondary and len(interaction_features) < 2:
        interaction_features.extend(
            summary_df[summary_df["is_important"] == "secondary"]["feature"].tolist()
        )

    # 3. 변수 쌍 생성 (자기 자신 제외)
    pairs = []

    for f in main_features:
        for inter in interaction_features:
            # 자기 자신과의 조합은 제외
            if f != inter:
                pairs.append((f, inter))

    # paris를 역순으로 순회하며 중복 제거
    seen = set()
    for i in range(len(pairs) - 1, -1, -1):
        key = frozenset(pairs[i])
        if key in seen:
            del pairs[i]
        else:
            seen.add(key)

    # 중요도 순 정렬 (주 변수 기준)
    importance_rank = {}

    for i, row in summary_df.iterrows():
        importance_rank[row["feature"]] = i

    pairs = sorted(pairs, key=lambda k: importance_rank.get(k[0], 999))

    # 4. dependence plot 일괄 생성
    for feature_name, interaction_name in pairs:
        shap.dependence_plot(
            feature_name,
            shap_values,
            x,
            interaction_index=interaction_name,
            show=False,
        )

        # SHAP figure 직접 제어
        fig = plt.gcf()
        fig.set_size_inches(width / 100, height / 100)
        fig.set_dpi(config.dpi)
        ax = fig.get_axes()[0]

        plt.xlabel(feature_name, fontsize=config.label_font_size)
        plt.ylabel(f"SHAP value for {feature_name}", fontsize=config.label_font_size)
        plt.xticks(fontsize=config.font_size)
        plt.yticks(fontsize=config.font_size)

        finalize_plot(
            ax,
            outparams=True,
            title=f"SHAP Dependence Plot: {feature_name} vs {interaction_name}",
        )

    return pairs

hs_describe 

hs_describe(data, *fields, columns=None)

데이터프레임의 연속형 변수의 단위 및 현실성을 평가하기 위해 확장된 기술통계량을 반환한다.

각 연속형(숫자형) 컬럼의 기술통계량(describe)을 구하고, 이에 사분위수 범위(IQR), 이상치 경계값(UP, DOWN), 왜도(skew), 이상치 개수 및 비율, 분포 특성, 로그변환 필요성을 추가하여 반환한다.

Parameters:

Name Type Description Default
data DataFrame

분석 대상 데이터프레임.

 required
*fields str

분석할 컬럼명 목록. 지정하지 않으면 모든 숫자형 컬럼을 처리.

 ()
columns list

반환할 통계량 컬럼 목록. None이면 모든 통계량 반환.

 None

Returns:

Name Type Description
DataFrame

각 필드별 확장된 기술통계량을 포함한 데이터프레임. 행은 다음과 같은 통계량을 포함:

  • count (float): 비결측치의 수
  • na_count (int): 결측치의 수
  • na_rate (float): 결측치 비율(%)
  • mean (float): 평균값
  • std (float): 표준편차
  • min (float): 최소값
  • 25% (float): 제1사분위수 (Q1)
  • 50% (float): 제2사분위수 (중앙값)
  • 75% (float): 제3사분위수 (Q3)
  • max (float): 최대값
  • iqr (float): 사분위 범위 (Q3 - Q1)
  • up (float): 이상치 상한 경계값 (Q3 + 1.5 * IQR)
  • down (float): 이상치 하한 경계값 (Q1 - 1.5 * IQR)
  • skew (float): 왜도
  • outlier_count (int): 이상치 개수
  • outlier_rate (float): 이상치 비율(%)
  • dist (str): 분포 특성 ("극단 우측 꼬리", "거의 대칭" 등)
  • log_need (str): 로그변환 필요성 ("높음", "중간", "낮음")

Examples:

from hossam import *
from pandas import DataFrame

df = DataFrame({
    'x': [1, 2, 3, 4, 5, 100],
    'y': [10, 20, 30, 40, 50, 60],
    'z': ['a', 'b', 'c', 'd', 'e', 'f']
})

# 전체 숫자형 컬럼에 대한 확장된 기술통계:
result = hs_stats.describe(df)
print(result)

# 특정 컬럼만 분석:
result = hs_stats.describe(df, 'x', 'y')
print(result)
Notes
  • 숫자형이 아닌 컬럼은 자동으로 제외됩니다.
  • 결과는 필드(컬럼)가 행으로, 통계량이 열로 구성됩니다.
  • Tukey의 1.5 * IQR 규칙을 사용하여 이상치를 판정합니다.
  • 분포 특성은 왜도 값으로 판정합니다.
  • 로그변환 필요성은 왜도의 절댓값 크기로 판정합니다.
Source code in hossam/hs_stats.py 
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
def describe(data: DataFrame, *fields: str, columns: list | None = None):
    """데이터프레임의 연속형 변수의 단위 및 현실성을 평가하기 위해 확장된 기술통계량을 반환한다.

    각 연속형(숫자형) 컬럼의 기술통계량(describe)을 구하고, 이에 사분위수 범위(IQR),
    이상치 경계값(UP, DOWN), 왜도(skew), 이상치 개수 및 비율, 분포 특성, 로그변환 필요성을
    추가하여 반환한다.

    Args:
        data (DataFrame): 분석 대상 데이터프레임.
        *fields (str): 분석할 컬럼명 목록. 지정하지 않으면 모든 숫자형 컬럼을 처리.
        columns (list, optional): 반환할 통계량 컬럼 목록. None이면 모든 통계량 반환.

    Returns:
        DataFrame: 각 필드별 확장된 기술통계량을 포함한 데이터프레임.
            행은 다음과 같은 통계량을 포함:

            - count (float): 비결측치의 수
            - na_count (int): 결측치의 수
            - na_rate (float): 결측치 비율(%)
            - mean (float): 평균값
            - std (float): 표준편차
            - min (float): 최소값
            - 25% (float): 제1사분위수 (Q1)
            - 50% (float): 제2사분위수 (중앙값)
            - 75% (float): 제3사분위수 (Q3)
            - max (float): 최대값
            - iqr (float): 사분위 범위 (Q3 - Q1)
            - up (float): 이상치 상한 경계값 (Q3 + 1.5 * IQR)
            - down (float): 이상치 하한 경계값 (Q1 - 1.5 * IQR)
            - skew (float): 왜도
            - outlier_count (int): 이상치 개수
            - outlier_rate (float): 이상치 비율(%)
            - dist (str): 분포 특성 ("극단 우측 꼬리", "거의 대칭" 등)
            - log_need (str): 로그변환 필요성 ("높음", "중간", "낮음")

    Examples:
        ```python
        from hossam import *
        from pandas import DataFrame

        df = DataFrame({
            'x': [1, 2, 3, 4, 5, 100],
            'y': [10, 20, 30, 40, 50, 60],
            'z': ['a', 'b', 'c', 'd', 'e', 'f']
        })

        # 전체 숫자형 컬럼에 대한 확장된 기술통계:
        result = hs_stats.describe(df)
        print(result)

        # 특정 컬럼만 분석:
        result = hs_stats.describe(df, 'x', 'y')
        print(result)
        ```

    Notes:
        - 숫자형이 아닌 컬럼은 자동으로 제외됩니다.
        - 결과는 필드(컬럼)가 행으로, 통계량이 열로 구성됩니다.
        - Tukey의 1.5 * IQR 규칙을 사용하여 이상치를 판정합니다.
        - 분포 특성은 왜도 값으로 판정합니다.
        - 로그변환 필요성은 왜도의 절댓값 크기로 판정합니다.
    """
    if columns is not None:
        if fields:  # type: ignore
            raise ValueError("fields와 columns 인자는 중복 사용할 수 없습니다.")
        fields = columns # type: ignore

    num_columns = list(data.select_dtypes(include=np.number).columns)

    target_fields: list | None = list(fields) if fields else num_columns # type: ignore

    # 기술통계량 구하기
    desc = data[target_fields].describe().T

    # 각 컬럼별 결측치 수(na_count) 추가
    na_counts = data[target_fields].isnull().sum()
    desc.insert(1, 'na_count', na_counts)

    # 결측치 비율(na_rate) 추가
    desc.insert(2, 'na_rate', (na_counts / len(data)) * 100)

    # 추가 통계량 계산
    additional_stats = []
    for f in target_fields: # type: ignore
        # 숫자 타입이 아니라면 건너뜀
        if f not in num_columns:
            continue

        # 사분위수
        q1 = data[f].quantile(q=0.25)
        q3 = data[f].quantile(q=0.75)

        # 이상치 경계 (Tukey's fences)
        iqr = q3 - q1
        down = q1 - 1.5 * iqr
        up = q3 + 1.5 * iqr

        # 왜도
        skew = data[f].skew()

        # 이상치 개수 및 비율
        outlier_count = ((data[f] < down) | (data[f] > up)).sum()
        outlier_rate = (outlier_count / len(data)) * 100

        # 분포 특성 판정 (왜도 기준)
        abs_skew = abs(skew)    # type: ignore
        if abs_skew < 0.5:      # type: ignore
            dist = "거의 대칭"
        elif abs_skew < 1.0:    # type: ignore
            if skew > 0:        # type: ignore
                dist = "약한 우측 꼬리"
            else:
                dist = "약한 좌측 꼬리"
        elif abs_skew < 2.0:    # type: ignore
            if skew > 0:        # type: ignore
                dist = "중간 우측 꼬리"
            else:
                dist = "중간 좌측 꼬리"
        else:
            if skew > 0:        # type: ignore
                dist = "극단 우측 꼬리"
            else:
                dist = "극단 좌측 꼬리"

        # 로그변환 필요성 판정
        if abs_skew < 0.5:      # type: ignore
            log_need = "낮음"
        elif abs_skew < 1.0:    # type: ignore
            log_need = "중간"
        else:
            log_need = "높음"

        additional_stats.append({
            'field': f,
            'iqr': iqr,
            'up': up,
            'down': down,
            'outlier_count': outlier_count,
            'outlier_rate': outlier_rate,
            'skew': skew,
            'dist': dist,
            'log_need': log_need
        })

    additional_df = DataFrame(additional_stats).set_index('field')

    # 결과 병합
    result = concat([desc, additional_df], axis=1)

    return result

hs_category_describe 

hs_category_describe(data, *fields, columns=None)

데이터프레임의 명목형(범주형) 변수에 대한 분포 편향을 요약한다.

각 명목형 컬럼의 최다 범주와 최소 범주의 정보를 요약하여 데이터프레임으로 반환한다.

Parameters:

Name Type Description Default
data DataFrame

분석 대상 데이터프레임.

 required
*fields str

분석할 컬럼명 목록. 지정하지 않으면 모든 명목형 컬럼을 처리.

 ()
columns list | None

분석할 컬럼명 목록. fields와 중복 사용 불가. 기본값은 None.

 None

Returns:

Type Description

tuple[DataFrame, DataFrame]: 각 컬럼별 최다/최소 범주 정보를 포함한 데이터프레임과 각 범주별 빈도/비율 정보를 포함한 데이터프레임을 튜플로 반환. 다음 컬럼을 포함:

  • 변수 (str): 컬럼명
  • 최다_범주: 가장 빈도가 높은 범주값
  • 최다_비율(%) (float): 최다 범주의 비율
  • 최소_범주: 가장 빈도가 낮은 범주값
  • 최소_비율(%) (float): 최소 범주의 비율

Examples:

from hossam import *
from pandas import DataFrame

df = DataFrame({
    'cut': ['Ideal', 'Premium', 'Good', 'Ideal', 'Premium'],
    'color': ['E', 'F', 'G', 'E', 'F'],
    'price': [100, 200, 150, 300, 120]
})

# 전체 명목형 컬럼에 대한 요약:
result, summary = hs_stats.category_describe(df)

# 특정 컬럼만 분석:
result, summary = hs_stats.category_describe(df, 'cut', 'color')
Notes
  • 숫자형 컬럼은 자동으로 제외됩니다.
  • NaN 값도 하나의 범주로 포함됩니다.
Source code in hossam/hs_stats.py 
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
def category_describe(data: DataFrame, *fields: str, columns: list | None = None):
    """데이터프레임의 명목형(범주형) 변수에 대한 분포 편향을 요약한다.

    각 명목형 컬럼의 최다 범주와 최소 범주의 정보를 요약하여 데이터프레임으로 반환한다.

    Args:
        data (DataFrame): 분석 대상 데이터프레임.
        *fields (str): 분석할 컬럼명 목록. 지정하지 않으면 모든 명목형 컬럼을 처리.
        columns (list | None, optional): 분석할 컬럼명 목록. fields와 중복 사용 불가. 기본값은 None.

    Returns:
        tuple[DataFrame, DataFrame]: 각 컬럼별 최다/최소 범주 정보를 포함한 데이터프레임과
            각 범주별 빈도/비율 정보를 포함한 데이터프레임을 튜플로 반환.
            다음 컬럼을 포함:

            - 변수 (str): 컬럼명
            - 최다_범주: 가장 빈도가 높은 범주값
            - 최다_비율(%) (float): 최다 범주의 비율
            - 최소_범주: 가장 빈도가 낮은 범주값
            - 최소_비율(%) (float): 최소 범주의 비율

    Examples:
        ```python
        from hossam import *
        from pandas import DataFrame

        df = DataFrame({
            'cut': ['Ideal', 'Premium', 'Good', 'Ideal', 'Premium'],
            'color': ['E', 'F', 'G', 'E', 'F'],
            'price': [100, 200, 150, 300, 120]
        })

        # 전체 명목형 컬럼에 대한 요약:
        result, summary = hs_stats.category_describe(df)

        # 특정 컬럼만 분석:
        result, summary = hs_stats.category_describe(df, 'cut', 'color')
        ```

    Notes:
        - 숫자형 컬럼은 자동으로 제외됩니다.
        - NaN 값도 하나의 범주로 포함됩니다.
    """
    # columns 인자가 있으면 fields보다 우선한다.
    if columns is not None:
        if fields:  # type: ignore
            raise ValueError("fields와 columns 인자는 중복 사용할 수 없습니다.")
        fields = columns # type: ignore

    num_columns = list(data.select_dtypes(include=np.number).columns)

    target_fields = list(fields) if fields else columns

    if not target_fields:
        # 명목형(범주형) 컬럼 선택: object, category, bool 타입
        target_fields = data.select_dtypes(include=['object', 'category', 'bool']).columns # type: ignore

    result = []
    summary = []
    for f in target_fields:
        # 숫자형 컬럼은 건너뜀
        if f in num_columns:
            continue

        # 각 범주값의 빈도수 계산 (NaN 포함)
        value_counts = data[f].value_counts(dropna=False)

        # 범주별 빈도/비율 정보 추가 (category_table 기능)
        for category, count in value_counts.items():
            rate = (count / len(data)) * 100
            result.append({
                "변수": f,
                "범주": category,
                "빈도": count,
                "비율(%)": round(rate, 2)
            })

        if len(value_counts) == 0:
            continue

        # 최다/최소 범주 정보 추가 (category_describe 기능)
        max_category = value_counts.index[0]
        max_count = value_counts.iloc[0]
        max_rate = (max_count / len(data)) * 100
        min_category = value_counts.index[-1]
        min_count = value_counts.iloc[-1]
        min_rate = (min_count / len(data)) * 100
        summary.append({
            "변수": f,
            "최다_범주": max_category,
            "최다_비율(%)": round(max_rate, 2),
            "최소_범주": min_category,
            "최소_비율(%)": round(min_rate, 2)
        })

    return DataFrame(result), DataFrame(summary).set_index("변수")