jussam.my_ts

jussam.my_ts

set_index

set_index(data, column, freq=None, sort=True)

날짜 컬럼을 인덱스로 지정하고 관측 간격을 명시한다.

Parameters:

Name	Type	Description	Default
data	DataFrame	날짜 컬럼을 포함하는 데이터프레임.	required
column	str	인덱스로 사용할 날짜 컬럼명.	required
freq	str	관측 간격 (예: "MS", "D", "W"). None이면 지정하지 않음 (기본값: None).	None
sort	bool	인덱스를 오름차순으로 정렬할지 여부 (기본값: True).	True

Returns:

Name	Type	Description
DataFrame		날짜 인덱스가 설정된 데이터프레임.

Source code in jussam/my_ts.py

def set_index(data, column, freq=None, sort=True):
    """날짜 컬럼을 인덱스로 지정하고 관측 간격을 명시한다.

    Args:
        data (DataFrame): 날짜 컬럼을 포함하는 데이터프레임.
        column (str): 인덱스로 사용할 날짜 컬럼명.
        freq (str): 관측 간격 (예: "MS", "D", "W"). None이면 지정하지 않음 (기본값: None).
        sort (bool): 인덱스를 오름차순으로 정렬할지 여부 (기본값: True).

    Returns:
        DataFrame: 날짜 인덱스가 설정된 데이터프레임.
    """
    df = data.copy()

    # --- 1) 날짜 컬럼을 datetime으로 변환 ---
    # 이미 datetime이면 그대로 통과하므로 타입을 미리 가려낼 필요가 없다.
    # (pandas 3부터 문자열 컬럼의 dtype이 object가 아니라 str이라, 타입을 가려내면
    #  변환이 그대로 건너뛰어지고 아래 asfreq()에서 값이 전부 NaN이 된다)
    df[column] = to_datetime(df[column])

    # --- 2) 인덱스로 지정 ---
    df = df.set_index(column)

    if sort:
        df = df.sort_index()

    # --- 3) 관측 간격 명시 ---
    # 간격을 명시해야 window·period 값이 "시간"을 뜻하게 된다.
    # 간격이 없으면 12는 "12개월"이 아니라 그냥 "관측치 12개"일 뿐이다.
    if freq is not None:
        df = df.asfreq(freq)

    return df

report_split

report_split(data, column=None, size=3)

시계열을 여러 구간으로 잘라 구간별 평균·표준편차·변동계수를 비교한다.

Parameters:

Name	Type	Description	Default
data	DataFrame | Series	날짜 인덱스를 가진 시계열 데이터.	required
column	str	data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).	None
size	int	나눌 구간의 개수 (기본값: 3).	3

Returns:

Name	Type	Description
DataFrame		구간별 시작·종료·관측치 수·평균·표준편차·변동계수.

Source code in jussam/my_ts.py

def report_split(data, column=None, size=3):
    """시계열을 여러 구간으로 잘라 구간별 평균·표준편차·변동계수를 비교한다.

    Args:
        data (DataFrame | Series): 날짜 인덱스를 가진 시계열 데이터.
        column (str): data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).
        size (int): 나눌 구간의 개수 (기본값: 3).

    Returns:
        DataFrame: 구간별 시작·종료·관측치 수·평균·표준편차·변동계수.
    """
    if column is not None:          # data가 데이터프레임이면 대상 컬럼만 추출
        data = data[column]

    length = len(data) // size      # 구간 하나의 길이
    result = []                     # 결과를 담을 리스트

    for i in range(size):
        # i번째 구간을 잘라낸다
        part = data.iloc[i * length : (i + 1) * length]

        result.append({
            "구간": f"구간 {i + 1}",
            "시작": part.index[0].strftime("%Y-%m-%d"),
            "종료": part.index[-1].strftime("%Y-%m-%d"),
            "관측치 수": len(part),
            "평균": round(part.mean(), 3),
            "표준편차": round(part.std(), 3),
            "변동계수": round(part.std() / part.mean(), 4),
        })

    return DataFrame(result).set_index("구간")

report_variation

report_variation(data, column=None, freq='YE')

기간 단위로 묶어 변동폭이 수준에 비례해 커지는지 확인한다.

표준편차는 커지는데 변동계수(표준편차/평균)가 일정하면 변동폭이 수준에 비례한다는 뜻이다. 이 경우 로그변환이 필요하다.

Parameters:

Name	Type	Description	Default
data	DataFrame | Series	날짜 인덱스를 가진 시계열 데이터.	required
column	str	data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).	None
freq	str	묶을 기간 단위 (기본값: "YE"). YE → 연말 기준 1년씩 / QE → 분기씩 / ME → 월말 기준 한 달씩, WE → 주말 기준 1주일씩 / D → 하루씩	'YE'

Returns:

Name	Type	Description
DataFrame		기간별 평균·표준편차·변동계수.

Source code in jussam/my_ts.py

def report_variation(data, column=None, freq="YE"):
    """기간 단위로 묶어 변동폭이 수준에 비례해 커지는지 확인한다.

    표준편차는 커지는데 변동계수(표준편차/평균)가 일정하면 변동폭이 수준에 비례한다는 뜻이다.
    이 경우 로그변환이 필요하다.

    Args:
        data (DataFrame | Series): 날짜 인덱스를 가진 시계열 데이터.
        column (str): data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).
        freq (str): 묶을 기간 단위 (기본값: "YE").
            YE → 연말 기준 1년씩  /  QE → 분기씩  /  ME → 월말 기준 한 달씩, 
            WE → 주말 기준 1주일씩  /  D → 하루씩

    Returns:
        DataFrame: 기간별 평균·표준편차·변동계수.
    """
    # --- 0) 대상 컬럼 추출 ---
    if column is not None:
        data = data[column]

    # --- 1) 기간 단위로 묶어 평균과 표준편차를 계산 ---
    # groupby()는 "값이 같은 것끼리" 묶는데, 날짜는 값이 전부 달라서 그대로는 묶이지 않는다.
    # Grouper(freq=...)를 키로 주면 "이 시간 간격으로 잘라서 묶어라"라는 뜻이 된다.
    #   freq="YE" → 연말 기준 1년씩  /  "QE" → 분기씩  /  "ME" → 월말 기준 한 달씩
    # 인덱스는 각 구간의 끝 날짜가 된다. (1949년 묶음 → 1949-12-31)
    group = data.groupby(Grouper(freq=freq))

    result_df = DataFrame({
        "평균": group.mean().round(3),
        "표준편차": group.std().round(3),
    })

    # --- 2) 변동계수 = 표준편차 / 평균 ---
    result_df["변동계수"] = (result_df["표준편차"] / result_df["평균"]).round(4)
    result_df = result_df.dropna()
    result_df.index.name = "기간"

    # --- 3) 최대/최소 비율로 판정 ---
    std_ratio = result_df["표준편차"].max() / result_df["표준편차"].min()
    cv_ratio = result_df["변동계수"].max() / result_df["변동계수"].min()

    if cv_ratio < std_ratio:
        conclusion = "변동폭이 수준에 비례한다 → 로그변환 권장 / 승법(multiplicative) 모델"
    else:
        conclusion = "변동폭이 일정하다 → 원본 사용 가능 / 가법(additive) 모델"

    print(f"표준편차 최대/최소: {std_ratio:.2f}배")
    print(f"변동계수 최대/최소: {cv_ratio:.2f}배")
    print(f"판정: {conclusion}")

    return result_df

adf_test

adf_test(data, column=None, name=None, alpha=0.05)

시계열 하나의 정상성을 ADF 검정으로 판정한다.

여러 대상을 비교하는 adf_diff · adf_transform 도 이 함수를 반복 호출한다. 반환값이 가로 1행 표이므로 concat() 으로 이어 붙이면 그대로 여러 행이 된다.

Parameters:

Name	Type	Description	Default
data	DataFrame | Series	검정할 시계열 데이터.	required
column	str	data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).	None
name	str	결과표에 표시할 행 이름 (기본값: None). 생략하면 컬럼명 → Series의 이름 → "시계열" 순으로 찾아 쓴다.	None
alpha	float	유의수준 (기본값: 0.05).	0.05

Returns:

Name	Type	Description
DataFrame		대상을 인덱스로 하는 가로 1행 ADF 검정 결과표.

Source code in jussam/my_ts.py

def adf_test(data, column=None, name=None, alpha=0.05):
    """시계열 하나의 정상성을 ADF 검정으로 판정한다.

    여러 대상을 비교하는 adf_diff · adf_transform 도 이 함수를 반복 호출한다.
    반환값이 가로 1행 표이므로 concat() 으로 이어 붙이면 그대로 여러 행이 된다.

    Args:
        data (DataFrame | Series): 검정할 시계열 데이터.
        column (str): data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).
        name (str): 결과표에 표시할 행 이름 (기본값: None).
            생략하면 컬럼명 → Series의 이름 → "시계열" 순으로 찾아 쓴다.
        alpha (float): 유의수준 (기본값: 0.05).

    Returns:
        DataFrame: 대상을 인덱스로 하는 가로 1행 ADF 검정 결과표.
    """
    # --- 1) 기본 준비 및 정상성 검정 ---
    if column is not None:              # 컬럼명이 있다면
        data = data[column]             # 대상 데이터만 추출

    # 표의 행 이름 가져오기
    if name is None:                    # 이름을 직접 주지 않았다면
        if column is not None:
            name = column               # 컬럼명을 행 이름으로 사용
        else:
            name = data.name            # Series의 이름을 행 이름으로 사용

    if name is None:  name = "시계열"    # 그래도 이름이 없으면 그냥 "시계열"로 지정

    x = Series(data).dropna()           # 결측치 제거

    # adfuller()는 결과를 튜플로 돌려준다
    statistic, pvalue, usedlag, nobs, cvalues, icbest = adfuller(x)

    # --- 2) 결과 구성 및 반환 ---
    stationary = bool(pvalue < alpha)   # p-value가 작아야 정상성 확보

    # 결과표 반환
    return DataFrame([{
        "관측치 수": len(x),
        "검정통계량(ADF)": round(statistic, 3),
        "p-value": round(pvalue, 4),
        "사용 시차": usedlag,
        "1% 기각값": round(cvalues["1%"], 3),
        "5% 기각값": round(cvalues["5%"], 3),
        "10% 기각값": round(cvalues["10%"], 3),
        "표준편차": round(x.std(), 3),
        "정상성": stationary,
        "판정": "정상" if stationary else "비정상",
    }], index=[name])

transform

transform(
    data,
    column=None,
    log=False,
    diff=0,
    seasonal_diff=0,
    period=None,
)

로그변환 · 일반차분 · 계절차분을 순서대로 적용한다.

Parameters:

Name	Type	Description	Default
data	DataFrame | Series	변환할 시계열 데이터.	required
column	str	data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).	None
log	bool	로그변환 적용 여부 (기본값: False).	False
diff	int	일반차분 횟수. ARIMA의 d에 해당 (기본값: 0).	0
seasonal_diff	int	계절차분 횟수. ARIMA의 D에 해당 (기본값: 0).	0
period	int	계절 주기. seasonal_diff가 1 이상이면 반드시 지정 (기본값: None).	None

Returns:

Name	Type	Description
Series		변환이 완료된 시계열. 차분으로 생긴 결측치는 제거된다.

Source code in jussam/my_ts.py

def transform(data, column=None, log=False, diff=0, seasonal_diff=0, period=None):
    """로그변환 · 일반차분 · 계절차분을 순서대로 적용한다.

    Args:
        data (DataFrame | Series): 변환할 시계열 데이터.
        column (str): data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).
        log (bool): 로그변환 적용 여부 (기본값: False).
        diff (int): 일반차분 횟수. ARIMA의 d에 해당 (기본값: 0).
        seasonal_diff (int): 계절차분 횟수. ARIMA의 D에 해당 (기본값: 0).
        period (int): 계절 주기. seasonal_diff가 1 이상이면 반드시 지정 (기본값: None).

    Returns:
        Series: 변환이 완료된 시계열. 차분으로 생긴 결측치는 제거된다.
    """
    # --- 0) 대상 컬럼 추출 ---
    if column is not None:
        data = data[column]

    result = data.copy()

    # --- 1) 로그변환 : 분산이 커지는 문제를 잡는다 ---
    if log:
        result = np.log(result)

    # --- 2) 일반차분 : 추세를 없앤다 ---
    for i in range(diff):
        result = result.diff()

    # --- 3) 계절차분 : 주기적 반복을 없앤다 ---
    for i in range(seasonal_diff):
        result = result.diff(period)

    return result.dropna()

adf_diff

adf_diff(data, column=None, max_diff=-1, alpha=0.05)

차분 차수를 늘려가며 ADF 검정을 수행하고 과대차분 여부를 판정한다.

표준편차가 최소가 되는 차수를 넘기면 데이터에 없던 노이즈를 만들어낸다. p-value를 낮추는 것이 목적이 아니라 필요한 만큼만 차분하는 것이 목적이다.

Parameters:

Name	Type	Description	Default
data	DataFrame | Series	검정할 시계열 데이터.	required
column	str	data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).	None
max_diff	int	최대 차분 횟수. -1이면 표준편차가 다시 커지는 지점, 즉 과대차분이 시작되는 곳에서 스스로 멈춘다 (기본값: -1).	-1
alpha	float	유의수준 (기본값: 0.05).	0.05

Returns:

Name	Type	Description
DataFrame		차수별 ADF 검정 결과와 권장 표시.

Source code in jussam/my_ts.py

def adf_diff(data, column=None, max_diff=-1, alpha=0.05):
    """차분 차수를 늘려가며 ADF 검정을 수행하고 과대차분 여부를 판정한다.

    표준편차가 최소가 되는 차수를 넘기면 데이터에 없던 노이즈를 만들어낸다.
    p-value를 낮추는 것이 목적이 아니라 필요한 만큼만 차분하는 것이 목적이다.

    Args:
        data (DataFrame | Series): 검정할 시계열 데이터.
        column (str): data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).
        max_diff (int): 최대 차분 횟수. -1이면 표준편차가 다시 커지는 지점,
            즉 과대차분이 시작되는 곳에서 스스로 멈춘다 (기본값: -1).
        alpha (float): 유의수준 (기본값: 0.05).

    Returns:
        DataFrame: 차수별 ADF 검정 결과와 권장 표시.
    """
    # --- 0) 대상 컬럼 추출 ---
    if column is not None:      # 컬럼 이름이 있다면
        data = data[column]     # 대상 데이터만 추출

    result = []                 # 결과를 담을 리스트
    target = data.copy()        # 검정할 대상 시계열. 차분할 때마다 관측치가 하나씩 줄어든다.

    order = 0                   # 지금까지 차분한 횟수
    previous_std = None         # 직전 차수의 표준편차
    stopped = None              # 과대차분이 검출된 차수 이름

    # --- 1) 0차(원본)부터 차수를 늘려가며 검정 ---
    while True:
        # --- 1-1) ADF 검정 ---
        name = "원본" if order == 0 else f"{order}차 차분"  # 차분 회차별 이름 구성
        row = adf_test(target, name=name, alpha=alpha)     # 검정 수행
        result.append(row)                                 # 검정 결과를 리스트에 추가
        current_std = row["표준편차"].iloc[0]               # 검정 결과에서 표준편차만 가져오기

        # --- 1-2) 멈출 때가 되었는지 확인 ---
        if max_diff < 0:            # 최대 차분 횟수를 지정하지 않았다면
            # --- 표준편차가 직전보다 커졌다면 이미 한 번 지나친 것이므로 여기서 멈춘다
            if previous_std is not None and current_std > previous_std:
                stopped = name      # --- 과대차분이 시작된 차수 이름 기록
                break               # --- 반복을 멈춘다
        elif order >= max_diff:     # 최대 차분 횟수를 지정했고 그 횟수에 도달했다면
            break                   # --- 반복을 멈춘다

        # --- 1-3) 다음 차수 준비 (반복을 멈춘 경우 아래 코드는 실행 안됨) ---
        previous_std = current_std  # 직전 차수의 표준편차를 현재 차수의 표준편차로 갱신
        order = order + 1           # 차분 횟수 1 증가

        # 다음 차수를 위해 1차 차분 수행. 차분할 때마다 관측치가 하나씩 줄어든다.
        target = target.diff().dropna()

        # 관측치가 너무 적어지면 검정 자체가 의미를 잃으므로 멈춘다
        if len(target) < 10:
            break

    result_df = concat(result)            # 결과 리스트를 데이터프레임으로 합치기

    # --- 2) 표준편차가 최소인 차수 확인 --> 과대차분 시작 직전 지점 ---
    best = result_df["표준편차"].idxmin()                   # 표준편차가 최소인 차수 이름
    best_at = list(result_df.index).index(best)            # 그 차수가 몇 번째 행인지
    recommend = []                                          # 권장 표시를 담을 리스트

    for position, name in enumerate(result_df.index):       # 각 행을 순서대로 순회하며
        if position == best_at:                             # 표준편차가 최소인 차수라면
            recommend.append("★ 여기까지")                   # 여기서 멈추라는 표시
        elif position > best_at:                            # 최소점을 지난 차수라면
            recommend.append("과대차분")                     # 차분이 지나쳤다는 표시
        else:                                               # 최소점 이전이라면
            recommend.append("")                            # 빈 문자열을 넣는다

    result_df["권장"] = recommend           # 권장 표시를 결과표에 추가

    # --- 3) 판단 근거를 글로 알린다 ---
    if stopped is not None:                 # 자동으로 멈춘 경우라면
        print(f"{stopped}에서 표준편차가 다시 커져 중단했습니다. (과대차분 시작)")

    print(f"권장 차분 차수: {best} (표준편차가 최소인 지점)")

    # 권장 차수가 아직 비정상이면 차분을 더 하는 것이 답이 아님을 알린다.
    # 남은 원인은 대개 계절성과 분산이며, 그것은 차분 횟수로 풀 문제가 아니다.
    if not result_df.loc[best, "정상성"]:
        pvalue = result_df.loc[best, "p-value"]
        print(f"⚠️ {best}은 아직 비정상입니다(p-value {pvalue}). 차분을 더 늘리지 말고 "
              f"로그변환·계절차분으로 남은 구조를 처리하세요. → adf_transform()")

    return result_df                        # 결과표 반환

adf_transform

adf_transform(
    data,
    column=None,
    period=None,
    max_diff=2,
    log=True,
    alpha=0.05,
)

로그변환 · 일반차분 · 계절차분의 조합을 만들어 전부 ADF 검정한다.

정상성을 만족하는 조합 중 표준편차가 가장 작은 것이 맨 위에 온다.

Parameters:

Name	Type	Description	Default
data	DataFrame | Series	검정할 시계열 데이터.	required
column	str	data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).	None
period	int	계절 주기. None이면 계절차분 조합을 만들지 않음 (기본값: None).	None
max_diff	int	조합에 포함할 최대 일반차분 횟수 (기본값: 2).	2
log	bool	로그변환 조합을 포함할지 여부 (기본값: True).	True
alpha	float	유의수준 (기본값: 0.05).	0.05

Returns:

Name	Type	Description
Series		권장 조합을 적용해 변환이 끝난 시계열. 조합 이름이 Series의 이름으로 붙는다. 조합별 검정 결과표는 함수 안에서 화면에 출력한다.

Source code in jussam/my_ts.py

def adf_transform(data, column=None, period=None, max_diff=2, log=True, alpha=0.05):
    """로그변환 · 일반차분 · 계절차분의 조합을 만들어 전부 ADF 검정한다.

    정상성을 만족하는 조합 중 표준편차가 가장 작은 것이 맨 위에 온다.

    Args:
        data (DataFrame | Series): 검정할 시계열 데이터.
        column (str): data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).
        period (int): 계절 주기. None이면 계절차분 조합을 만들지 않음 (기본값: None).
        max_diff (int): 조합에 포함할 최대 일반차분 횟수 (기본값: 2).
        log (bool): 로그변환 조합을 포함할지 여부 (기본값: True).
        alpha (float): 유의수준 (기본값: 0.05).

    Returns:
        Series: 권장 조합을 적용해 변환이 끝난 시계열. 조합 이름이 Series의 이름으로 붙는다.
            조합별 검정 결과표는 함수 안에서 화면에 출력한다.
    """
    # --- 1) 조합 후보 정의 ---
    if column is not None:          # 컬럼 이름이 있다면
        data = data[column]         # 대상 데이터만 추출

    log_options = [False]           # 로그변환 여부 후보. False만 넣어두고 True를 추가할지 결정

    if log:                         # 사용자가 로그변환 조합을 포함하라고 했다면
        log_options.append(True)    # True를 후보에 추가

    seasonal_options = [0]          # 주기를 모르면 계절차분은 시도하지 않는다

    if period is not None:          # 주기를 지정했다면
        seasonal_options.append(1)  # 계절차분 1회 조합을 후보에 추가

    result = []                     # 검정 결과를 담을 리스트
    changed = {}                    # 조합 이름별 변환 결과를 담을 딕셔너리

    # --- 2) 모든 조합을 순회하며 변환 후 검정 ---
    for use_log in log_options:                     # 로그변환 여부 후보 순회
        for diff in range(max_diff + 1):            # 일반차분 횟수 후보 순회
            for seasonal_diff in seasonal_options:  # 계절차분 여부 후보 순회

                # --- 2-1) 조합 이름 구성 ---
                parts = []
                if use_log:                     # 로그변환을 적용했다면 조합이름에 추가
                    parts.append("로그변환")

                if diff > 0:                    # 일반차분을 적용했다면 조합이름에 추가
                    parts.append(f"{diff}차 차분") 

                if seasonal_diff > 0:           # 계절차분을 적용했다면 조합이름에 추가
                    parts.append(f"계절차분({period})")

                if len(parts) == 0:             # 조합이름이 비어있다면 원본임
                    name = "원본"
                else:                           # 조합이름이 있다면 +로 이어붙인다
                    name = " + ".join(parts)

                # --- 2-2) 조합대로 변환 후 ADF 검정 ---
                target = data.copy()            # 조합대로 변환을 수행하기 위해 원본 복사

                if use_log:                     # 로그변환을 적용했다면 처리한다.
                    target = np.log(target)

                for i in range(diff):           # 일반차분을 적용했다면 처리한다.
                    target = target.diff()

                for i in range(seasonal_diff):  # 계절차분을 적용했다면 처리한다.
                    target = target.diff(period)

                target = target.dropna()        # 차분으로 생긴 결측치 제거
                target.name = name              # 변환 결과에 조합 이름을 붙인다

                changed[name] = target          # 나중에 골라 돌려주기 위해 보관
                result.append(adf_test(target, name=name, alpha=alpha))

    # --- 3) 정상성 우선, 같으면 표준편차가 작은 순으로 정렬 ---
    result_df = concat(result)
    result_df = result_df.sort_values(["정상성", "표준편차"], ascending=[False, True])

    best = result_df.index[0]
    best_std = result_df["표준편차"].iloc[0]

    print(f"조합 {len(result_df)}개 중 정상 {int(result_df['정상성'].sum())}개")
    print(f"권장 전처리: {best}")

    # --- 4) 과대차분 경고 ---
    # 권장한 조합보다 표준편차가 더 작은데 정상성만 못 갖춘 조합이 있다면,
    # 차분을 한 번 더 해서 정상성을 얻는 대신 없던 노이즈를 만들었을 수 있다.
    smaller = result_df[result_df["표준편차"] < best_std]

    if len(smaller) > 0:
        name = smaller["표준편차"].idxmin()
        std = smaller["표준편차"].min()
        print(f"⚠️ {name}의 표준편차({std})가 더 작습니다. "
              f"과대차분일 수 있으니 계절차분·로그변환을 함께 검토하세요.")

    display(result_df)      # 결과 표 출력

    return changed[best]    # 최적의 전처리 결과 반환

plot_rolling

plot_rolling(
    data,
    window,
    column=None,
    title=None,
    xlabel=None,
    ylabel=None,
    width=1280,
    height=480,
    save_path=None,
)

원본과 이동평균·이동표준편차를 한 그래프에 그려 정상성을 눈으로 확인한다.

정상 시계열이라면 이동평균과 이동표준편차가 모두 수평선에 가까워야 한다.

Parameters:

Name	Type	Description	Default
data	DataFrame | Series	날짜 인덱스를 가진 시계열 데이터.	required
window	int	이동 계산에 사용할 창 크기. 계절 주기로 잡는다.	required
column	str	data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).	None
title	str	그래프 제목 (기본값: None).	None
xlabel	str	x축 레이블 (기본값: None).	None
ylabel	str	y축 레이블 (기본값: None).	None
width	int	캔버스 가로 픽셀 (기본값: 1280).	1280
height	int	캔버스 세로 픽셀 (기본값: 480).	480
save_path	str	이미지 저장 경로 (기본값: None).	None

Returns:

Name	Type	Description
DataFrame		원본·이동평균·이동표준편차.

Source code in jussam/my_ts.py

def plot_rolling(data, window, column=None, title=None, xlabel=None, ylabel=None,
                 width=1280, height=480, save_path=None):
    """원본과 이동평균·이동표준편차를 한 그래프에 그려 정상성을 눈으로 확인한다.

    정상 시계열이라면 이동평균과 이동표준편차가 모두 수평선에 가까워야 한다.

    Args:
        data (DataFrame | Series): 날짜 인덱스를 가진 시계열 데이터.
        window (int): 이동 계산에 사용할 창 크기. 계절 주기로 잡는다.
        column (str): data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).
        title (str): 그래프 제목 (기본값: None).
        xlabel (str): x축 레이블 (기본값: None).
        ylabel (str): y축 레이블 (기본값: None).
        width (int): 캔버스 가로 픽셀 (기본값: 1280).
        height (int): 캔버스 세로 픽셀 (기본값: 480).
        save_path (str): 이미지 저장 경로 (기본값: None).

    Returns:
        DataFrame: 원본·이동평균·이동표준편차.
    """
    # --- 1) 대상 데이터 준비 ---
    if column is not None:      # 컬럼명이 전달되었다면
        data = data[column]     # 대상 컬럼만 추출

    # --- 2) 이동평균과 이동표준편차 계산 ---
    result_df = DataFrame({
        "원본": data,
        "이동평균": data.rolling(window).mean(),
        "이동표준편차": data.rolling(window).std(),
    })

    # --- 3) 그래프 그리기 및 결과 반환 ---
    if title is None:
        title = f"원본 · 이동평균 · 이동표준편차 (window={window})"

    fig, ax = my_plot.init(width=width, height=height, title=title,
                           xlabel=xlabel, ylabel=ylabel)

    for col in result_df.columns:
        my_plot.lineplot(x=result_df.index, y=result_df[col], label=col, ax=ax)

    ax.legend()
    my_plot.show(save_path=save_path)

    return result_df

compare_smoothing

compare_smoothing(
    data,
    sizes,
    column=None,
    method="ma",
    overlay=False,
    plot=True,
    title=None,
    xlabel=None,
    ylabel=None,
    width=1280,
    height=480,
    save_path=None,
)

창 크기(또는 span)를 바꿔가며 평활 결과를 비교한다.

Parameters:

Name	Type	Description	Default
data	DataFrame | Series	날짜 인덱스를 가진 시계열 데이터.	required
sizes	list | tuple	비교할 창 크기(또는 span) 목록.	required
column	str	data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).	None
method	str	"ma"(이동평균) · "ewm"(지수평활) · "both"(둘 다) (기본값: "ma").	'ma'
overlay	bool	True면 한 축에 겹쳐 그리고, False면 격자에 나눠 그린다 (기본값: False).	False
plot	bool	그래프를 그릴지 여부 (기본값: True).	True
title	str	그래프 제목 (기본값: None).	None
xlabel	str	x축 레이블 (기본값: None).	None
ylabel	str	y축 레이블 (기본값: None).	None
width	int	캔버스 가로 픽셀 (기본값: 960).	1280
height	int	캔버스 세로 픽셀 (기본값: 400).	480
save_path	str	이미지 저장 경로 (기본값: None).	None

Returns:

Name	Type	Description
DataFrame		원본과 평활 결과.

Source code in jussam/my_ts.py

def compare_smoothing(data, sizes, column=None, method="ma", overlay=False,
                      plot=True, title=None, xlabel=None, ylabel=None,
                      width=1280, height=480, save_path=None):
    """창 크기(또는 span)를 바꿔가며 평활 결과를 비교한다.

    Args:
        data (DataFrame | Series): 날짜 인덱스를 가진 시계열 데이터.
        sizes (list | tuple): 비교할 창 크기(또는 span) 목록.
        column (str): data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).
        method (str): "ma"(이동평균) · "ewm"(지수평활) · "both"(둘 다) (기본값: "ma").
        overlay (bool): True면 한 축에 겹쳐 그리고, False면 격자에 나눠 그린다 (기본값: False).
        plot (bool): 그래프를 그릴지 여부 (기본값: True).
        title (str): 그래프 제목 (기본값: None).
        xlabel (str): x축 레이블 (기본값: None).
        ylabel (str): y축 레이블 (기본값: None).
        width (int): 캔버스 가로 픽셀 (기본값: 960).
        height (int): 캔버스 세로 픽셀 (기본값: 400).
        save_path (str): 이미지 저장 경로 (기본값: None).

    Returns:
        DataFrame: 원본과 평활 결과.
    """
    # --- 1) 창 크기별 평활 계산 ---
    if column is not None:                  # 컬럼명이 전달되었다면
        data = data[column]                 # 대상 컬럼만 추출

    result_df = DataFrame({"원본": data})   # 결과를 담을 데이터프레임 준비

    for size in sizes:                      # 각 창 크기별로 평활 계산
        # --- 1-1) 이동평균 ---
        if method == "ma" or method == "both":
            result_df[f"이동평균({size})"] = data.rolling(size).mean()

        # --- 1-2) 지수평활 ---
        if method == "ewm" or method == "both":
            # ewm()의 첫 인자는 span이 아니라 com이므로 span을 키워드로 명시한다
            result_df[f"지수평활({size})"] = data.ewm(span=size).mean()

    # 그래프를 그리지 않으면 결과표만 반환하고 여기서 중단
    if not plot:
        return result_df

    # --- 2) 한 축에 겹쳐 그리기 ---
    if overlay:
        fig, ax = my_plot.init(width=width, height=height, title=title,
                               xlabel=xlabel, ylabel=ylabel)

        for col in result_df.columns:
            my_plot.lineplot(x=result_df.index, y=result_df[col], label=col, ax=ax)

        ax.legend()
        my_plot.show(save_path=save_path)

        # 겹쳐 그린 경우에도 결과표를 반환하고 여기서 중단
        return result_df

    # --- 3) 격자에 한 칸씩 그리기 ---
    cols = 2
    rows = int(np.ceil(len(result_df.columns) / cols))

    fig, ax = my_plot.init(rows=rows, cols=cols, width=width, height=height, title=title)

    for i, col in enumerate(result_df.columns):
        my_plot.lineplot(x=result_df.index, y=result_df[col], ax=ax[i])
        ax[i].set_title(col)

    # 컬럼 수가 홀수면 마지막 칸이 남으므로 숨긴다
    for i in range(len(result_df.columns), rows * cols):
        ax[i].set_visible(False)

    my_plot.show(save_path=save_path)

    return result_df

decompose

decompose(
    data,
    period,
    column=None,
    model=None,
    plot=True,
    title=None,
    rows=2,
    cols=2,
    width=960,
    height=480,
    save_path=None,
)

시계열을 추세 · 계절 · 잔차 세 성분으로 분해한다.

분해는 차분하기 전의 원본으로 수행한다. 차분은 추세와 계절성을 지우는 작업인데 분해는 그 추세와 계절성을 보려는 것이기 때문이다.

Parameters:

Name	Type	Description	Default
data	DataFrame | Series	날짜 인덱스를 가진 시계열 데이터.	required
period	int	계절 주기.	required
column	str	data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).	None
model	str	"additive" 또는 "multiplicative". None이면 자동 판정 (기본값: None).	None
plot	bool	성분별 그래프를 그릴지 여부 (기본값: True).	True
title	str	그래프 제목 (기본값: None).	None
rows	int	격자 행 수 (기본값: 2).	2
cols	int	격자 열 수 (기본값: 2).	2
width	int	캔버스 가로 픽셀 (기본값: 960).	960
height	int	한 단의 세로 픽셀 (기본값: 480).	480
save_path	str	이미지 저장 경로 (기본값: None).	None

Returns:

Name	Type	Description
DataFrame		원본·추세·계절·잔차 네 성분.

Source code in jussam/my_ts.py

def decompose(data, period, column=None, model=None, plot=True, title=None,
              rows=2, cols=2, width=960, height=480, save_path=None):
    """시계열을 추세 · 계절 · 잔차 세 성분으로 분해한다.

    분해는 차분하기 전의 원본으로 수행한다. 차분은 추세와 계절성을 지우는 작업인데
    분해는 그 추세와 계절성을 보려는 것이기 때문이다.

    Args:
        data (DataFrame | Series): 날짜 인덱스를 가진 시계열 데이터.
        period (int): 계절 주기.
        column (str): data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).
        model (str): "additive" 또는 "multiplicative". None이면 자동 판정 (기본값: None).
        plot (bool): 성분별 그래프를 그릴지 여부 (기본값: True).
        title (str): 그래프 제목 (기본값: None).
        rows (int): 격자 행 수 (기본값: 2).
        cols (int): 격자 열 수 (기본값: 2).
        width (int): 캔버스 가로 픽셀 (기본값: 960).
        height (int): 한 단의 세로 픽셀 (기본값: 480).
        save_path (str): 이미지 저장 경로 (기본값: None).

    Returns:
        DataFrame: 원본·추세·계절·잔차 네 성분.
    """

    # --- 1) 결합 방식 자동 판정 ---
    if column is not None:
        data = data[column]

    # 주기 단위로 묶어, 표준편차보다 변동계수가 더 일정하면 승법으로 본다
    if model is None:
        block = np.arange(len(data)) // period      # 주기 단위로 묶어 블록 번호를 만든다
        block_std = data.groupby(block).std()       # 각 블록별 표준편차
        block_mean = data.groupby(block).mean()     # 각 블록별 평균
        block_cv = block_std / block_mean           # 각 블록별 변동계수

        # 변동계수가 일정하면 승법, 그렇지 않으면 가법
        if (block_cv.max() / block_cv.min()) < (block_std.max() / block_std.min()):
            model = "multiplicative"    # 변동계수가 일정 → 승법
        else:
            model = "additive"          # 변동계수가 일정하지 않음 → 가법

        print(f"결합 방식 자동 판정: {model}")
    else:
        print(f"결합 방식 지정: {model}")

    # --- 2) 분해 수행 ---
    result = seasonal_decompose(data, model=model, period=period)

    result_df = DataFrame({
        "원본": result.observed,
        "추세": result.trend,
        "계절": result.seasonal,
        "잔차": result.resid,
    })

    print(f"관측치 {len(result_df)}개 중 추세·잔차가 계산된 구간 {len(result_df.dropna())}개")

    # --- 3) 성분별로 한 단씩 그리기 ---
    if plot:
        # --- 3-1) 그래프 초기화 ---
        if title is None:
            title = f"시계열 분해 ({model}, period={period})"

        fig, ax = my_plot.init(rows=2, cols=2, width=width, height=height, title=title)

        # --- 3-2) 각 성분별 시각화 ---
        for i, col in enumerate(result_df.columns):
            if col == "잔차":
                # 잔차는 불규칙하므로 점으로 그려야 남은 패턴이 보인다
                my_plot.scatterplot(data=None, x=result_df.index, y=result_df[col],
                                    size=8, linewidth=0.5, ax=ax[i])
            else:
                my_plot.lineplot(x=result_df.index, y=result_df[col], ax=ax[i])

            ax[i].set_title(col)
            ax[i].set_ylabel("")

        # --- 3-3) 잔차 기준선 그리기 ---
        # 잔차의 기준선은 승법이면 1, 가법이면 0이다
        base = 1 if model == "multiplicative" else 0

        # 잔차 그래프에 기준선 그리기
        ax[3].axhline(base, color="red", linestyle="--", linewidth=1)

        # --- 3-4) 그래프 출력 ---
        my_plot.show(save_path=save_path)

    # 결과 반환
    return result_df 

report_seasonal

report_seasonal(
    data,
    period,
    column=None,
    model=None,
    plot=True,
    title=None,
    xlabel=None,
    ylabel=None,
    width=1000,
    height=400,
    save_path=None,
)

계절 성분을 주기 안의 위치별로 정리해 계절지수표를 만든다.

승법 모델의 계절지수는 배수로 읽는다. 1.227은 추세선보다 22.7% 높다는 뜻이다.

Parameters:

Name	Type	Description	Default
data	DataFrame | Series	날짜 인덱스를 가진 시계열 데이터.	required
period	int	계절 주기.	required
column	str	data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).	None
model	str	"additive" 또는 "multiplicative". None이면 자동 판정 (기본값: None).	None
plot	bool	계절지수 그래프를 그릴지 여부 (기본값: True).	True
title	str	그래프 제목 (기본값: None).	None
xlabel	str	x축 레이블 (기본값: None).	None
ylabel	str	y축 레이블 (기본값: None).	None
width	int	캔버스 가로 픽셀 (기본값: 1000).	1000
height	int	캔버스 세로 픽셀 (기본값: 400).	400
save_path	str	이미지 저장 경로 (기본값: None).	None

Returns:

Name	Type	Description
DataFrame		주기 내 위치(1~period)를 인덱스로 하는 계절지수표.

Source code in jussam/my_ts.py

def report_seasonal(data, period, column=None, model=None, plot=True,
                    title=None, xlabel=None, ylabel=None,
                    width=1000, height=400, save_path=None):
    """계절 성분을 주기 안의 위치별로 정리해 계절지수표를 만든다.

    승법 모델의 계절지수는 배수로 읽는다. 1.227은 추세선보다 22.7% 높다는 뜻이다.

    Args:
        data (DataFrame | Series): 날짜 인덱스를 가진 시계열 데이터.
        period (int): 계절 주기.
        column (str): data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).
        model (str): "additive" 또는 "multiplicative". None이면 자동 판정 (기본값: None).
        plot (bool): 계절지수 그래프를 그릴지 여부 (기본값: True).
        title (str): 그래프 제목 (기본값: None).
        xlabel (str): x축 레이블 (기본값: None).
        ylabel (str): y축 레이블 (기본값: None).
        width (int): 캔버스 가로 픽셀 (기본값: 1000).
        height (int): 캔버스 세로 픽셀 (기본값: 400).
        save_path (str): 이미지 저장 경로 (기본값: None).

    Returns:
        DataFrame: 주기 내 위치(1~period)를 인덱스로 하는 계절지수표.
    """
    # --- 1) 결합 방식 자동 판정 ---
    if column is not None:
        data = data[column]

    if model is None:                               # 사용할 모델이 지정되지 않았다면?
        block = np.arange(len(data)) // period      # 주기 단위로 묶어 블록 번호를 만든다
        block_std = data.groupby(block).std()       # 각 블록별 표준편차
        block_mean = data.groupby(block).mean()     # 각 블록별 평균
        block_cv = block_std / block_mean           # 각 블록별 변동계수

        # 변동계수가 일정하면 승법, 그렇지 않으면 가법
        if (block_cv.max() / block_cv.min()) < (block_std.max() / block_std.min()):
            model = "multiplicative"
        else:
            model = "additive"

    # --- 2) 분해 후 계절 성분만 사용 ---
    # 분해 시계열 생성
    result = seasonal_decompose(data, model=model, period=period)

    # 계절 성분만 추출
    season = result.seasonal

    # 계절 성분은 매 주기 똑같이 반복되므로 한 주기만 보면 전체를 알 수 있다.
    # 위치는 데이터 시작 시점을 1로 하여 주기 안에서 센다.
    position = np.arange(len(season)) % period + 1
    index = season.groupby(position).first()

    result_df = DataFrame({"계절지수": index.round(4), 
                           "평균": index.mean().round(4)})
    result_df.index.name = "주기 내 위치"

    # --- 3) 기준 대비 크기 ---
    # 승법은 1이 기준이므로 백분율로, 가법은 0이 기준이므로 절대값으로 읽는다
    if model == "multiplicative":
        base = 1
        result_df["기준 대비(%)"] = ((index - 1) * 100).round(1)
    else:
        base = 0
        result_df["기준 대비"] = index.round(3)

    print(f"결합 방식: {model}")
    print(f"최고 위치: {index.idxmax()} ({index.max():.4f})")
    print(f"최저 위치: {index.idxmin()} ({index.min():.4f})")

    # --- 4) 그래프 그리기 ---
    if plot:
        if title is None:
            title = f"주기 내 위치별 계절지수 ({model}, period={period})"

        fig, ax = my_plot.init(width=width, height=height, title=title,
                               xlabel=xlabel, ylabel=ylabel)

        my_plot.lineplot(x=index.index, y=index.values, marker="o", ax=ax)
        ax.axhline(base, color="red", linestyle="--", linewidth=1)

        my_plot.show(save_path=save_path)

    # 결과 반환
    return result_df

adjust_seasonal

adjust_seasonal(
    data,
    period,
    column=None,
    model=None,
    plot=True,
    title=None,
    xlabel=None,
    ylabel=None,
    width=1280,
    height=480,
    save_path=None,
)

원본에서 계절 성분을 걷어내 계절조정 시계열을 만든다.

계절조정 결과에서 갑자기 튀는 지점은 계절성으로 설명되지 않는 사건이 있었다는 뜻이다.

Parameters:

Name	Type	Description	Default
data	DataFrame | Series	날짜 인덱스를 가진 시계열 데이터.	required
period	int	계절 주기.	required
column	str	data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).	None
model	str	"additive" 또는 "multiplicative". None이면 자동 판정 (기본값: None).	None
plot	bool	원본과 겹쳐 그릴지 여부 (기본값: True).	True
title	str	그래프 제목 (기본값: None).	None
xlabel	str	x축 레이블 (기본값: None).	None
ylabel	str	y축 레이블 (기본값: None).	None
width	int	캔버스 가로 픽셀 (기본값: 1280).	1280
height	int	캔버스 세로 픽셀 (기본값: 480).	480
save_path	str	이미지 저장 경로 (기본값: None).	None

Returns:

Name	Type	Description
Series		계절 요인이 제거된 시계열.

Source code in jussam/my_ts.py

def adjust_seasonal(data, period, column=None, model=None, plot=True, title=None,
                    xlabel=None, ylabel=None, width=1280, height=480, save_path=None):
    """원본에서 계절 성분을 걷어내 계절조정 시계열을 만든다.

    계절조정 결과에서 갑자기 튀는 지점은 계절성으로 설명되지 않는 사건이 있었다는 뜻이다.

    Args:
        data (DataFrame | Series): 날짜 인덱스를 가진 시계열 데이터.
        period (int): 계절 주기.
        column (str): data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).
        model (str): "additive" 또는 "multiplicative". None이면 자동 판정 (기본값: None).
        plot (bool): 원본과 겹쳐 그릴지 여부 (기본값: True).
        title (str): 그래프 제목 (기본값: None).
        xlabel (str): x축 레이블 (기본값: None).
        ylabel (str): y축 레이블 (기본값: None).
        width (int): 캔버스 가로 픽셀 (기본값: 1280).
        height (int): 캔버스 세로 픽셀 (기본값: 480).
        save_path (str): 이미지 저장 경로 (기본값: None).

    Returns:
        Series: 계절 요인이 제거된 시계열.
    """
    # --- 1) 결합 방식 자동 판정 ---
    if column is not None:
        data = data[column]

    if model is None:                               # 사용할 모델이 지정되지 않았다면?  
        block = np.arange(len(data)) // period      # 주기 단위로 묶어 블록 번호를 만든다
        block_std = data.groupby(block).std()       # 각 블록별 표준편차
        block_mean = data.groupby(block).mean()     # 각 블록별 평균
        block_cv = block_std / block_mean           # 각 블록별 변동계수

        # 변동계수가 일정하면 승법, 그렇지 않으면 가법
        if (block_cv.max() / block_cv.min()) < (block_std.max() / block_std.min()):
            model = "multiplicative"
        else:
            model = "additive"

    # --- 2) 계절 성분 제거 ---
    result = seasonal_decompose(data, model=model, period=period)

    # 승법이면 나누고, 가법이면 뺀다
    if model == "multiplicative":
        adjusted = result.observed / result.seasonal
        print(f"결합 방식: {model} (원본 ÷ 계절)")
    else:
        adjusted = result.observed - result.seasonal
        print(f"결합 방식: {model} (원본 − 계절)")

    # --- 3) 그래프 그리기 ---
    if plot:
        if title is None:
            title = f"원본 vs 계절조정 ({model}, period={period})"

        fig, ax = my_plot.init(width=width, height=height, title=title,
                               xlabel=xlabel, ylabel=ylabel)

        my_plot.lineplot(x=data.index, y=data, label="원본", ax=ax)
        my_plot.lineplot(x=adjusted.index, y=adjusted, label="계절조정", ax=ax)

        ax.legend()
        my_plot.show(save_path=save_path)

    # 결과 반환
    return adjusted

auto_correlation

auto_correlation(
    data,
    column=None,
    period=None,
    plot=True,
    rows=2,
    cols=1,
    width=1280,
    height=380,
    title=None,
    save_path=None,
)

ACF와 PACF를 한 번에 계산해 유의성과 차수 후보(p·q·P·Q)를 판정한다.

Parameters:

Name	Type	Description	Default
data	DataFrame | Series	정상성을 만족하는 시계열 데이터.	required
column	str	data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).	None
period	int	계절 주기. None이면 계절 차수를 판정하지 않음 (기본값: None).	None
plot	bool	ACF·PACF 그래프를 그릴지 여부 (기본값: True).	True
rows	int	그래프 격자의 행 수 (기본값: 2).	2
cols	int	그래프 격자의 열 수 (기본값: 1).	1
width	int	캔버스 가로 픽셀 (기본값: 1280).	1280
height	int	한 칸의 세로 픽셀 (기본값: 380).	380
title	str	그래프 제목 (기본값: None).	None
save_path	str	이미지 저장 경로 (기본값: None).	None

Returns:

Name	Type	Description
DataFrame		lag별 acf·pacf·유의성과 차수 위치 표시. 가장 뒤쪽 차수의 lag까지만 잘라서 반환하며, 차수가 모두 0이면 근거를 볼 수 있도록 전체를 반환한다.

Source code in jussam/my_ts.py

def auto_correlation(data, column=None, period=None, plot=True,
                       rows=2, cols=1, width=1280, height=380,
                       title=None, save_path=None):
    """ACF와 PACF를 한 번에 계산해 유의성과 차수 후보(p·q·P·Q)를 판정한다.

    Args:
        data (DataFrame | Series): 정상성을 만족하는 시계열 데이터.
        column (str): data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).
        period (int): 계절 주기. None이면 계절 차수를 판정하지 않음 (기본값: None).
        plot (bool): ACF·PACF 그래프를 그릴지 여부 (기본값: True).
        rows (int): 그래프 격자의 행 수 (기본값: 2).
        cols (int): 그래프 격자의 열 수 (기본값: 1).
        width (int): 캔버스 가로 픽셀 (기본값: 1280).
        height (int): 한 칸의 세로 픽셀 (기본값: 380).
        title (str): 그래프 제목 (기본값: None).
        save_path (str): 이미지 저장 경로 (기본값: None).

    Returns:
        DataFrame: lag별 acf·pacf·유의성과 차수 위치 표시.
            가장 뒤쪽 차수의 lag까지만 잘라서 반환하며,
            차수가 모두 0이면 근거를 볼 수 있도록 전체를 반환한다.
    """
    # --- 1) 대상 추출 ---
    if column is not None:
        data = data[column]

    data = data.dropna()
    n = len(data)

    # --- 2) 최대 시차 결정 ---
    # 상한: 뒤쪽 시차일수록 계산에 쓰이는 쌍이 줄어 추정이 불안정해진다.
    limit = n // 4

    if period is None:
        lags = min(int(10 * np.log10(n)), limit)    # statsmodels 기본값
    else:
        # 계절 주기의 3배를 노리되, 판정에 꼭 필요한 2배는 상한보다 우선한다
        lags = max(min(3 * period, limit), 2 * period)

    # --- 3) 계수 계산 (lag 0은 자기 자신이므로 제외) ---
    acf_values = acf(data, nlags=lags)[1:]
    pacf_values = pacf(data, nlags=lags)[1:]

    # 유의성 기준값: 2 / sqrt(N). 관측치가 많을수록 작아진다.
    threshold = 2 / np.sqrt(n)

    # --- 4) 결과표 구성 ---
    result_df = DataFrame({
        "lag": range(1, lags + 1),
        "acf": acf_values.round(3),
        "acf 유의성": np.abs(acf_values) > threshold,
        "pacf": pacf_values.round(3),
        "pacf 유의성": np.abs(pacf_values) > threshold,
        "기준값": round(threshold, 3),
    }).set_index("lag")

    # --- 5) 차수 판정용 내부 함수 정의 ---
    # 절단 지점은 "앞에서부터 유의하던 것이 처음으로 끊기는" 자리다.
    # 그 직전까지 연속으로 유의한 개수가 곧 차수가 된다.
    # 고립되어 튀어나온 유의성은 우연일 뿐이므로 차수로 세지 않는다.
    def __cut_off(significant):
        order = 0

        for sig in significant:
            if not sig:                 # 비유의를 만나는 순간이 절단 지점
                break

            order += 1                  # 여기까지는 연속으로 유의했다

        return order

    # --- 6) 차수 판정 ---
    # 비계절 차수: 이웃한 시차(1, 2, 3 …)를 순서대로 본다
    # ACF의 절단 → MA 차수(q), PACF의 절단 → AR 차수(p)
    q = __cut_off(result_df["acf 유의성"])
    p = __cut_off(result_df["pacf 유의성"])

    # 계절 차수: 주기의 배수(period, 2*period …)만 골라 같은 규칙을 적용한다
    # 세는 단위가 1시차에서 1주기로 바뀔 뿐이다
    season_lags = list(range(period, lags + 1, period)) if period else []

    Q = __cut_off(result_df.loc[season_lags, "acf 유의성"]) if season_lags else 0
    P = __cut_off(result_df.loc[season_lags, "pacf 유의성"]) if season_lags else 0

    # --- 7) 차수 위치 표시 ---
    result_df["p"] = np.where(result_df.index == p, "◀", "")
    result_df["q"] = np.where(result_df.index == q, "◀", "")

    if period is not None:
        # 계절 차수 N은 lag N*period 에서 확정된다 (차수 0이면 표시할 자리가 없다)
        result_df["P"] = np.where(result_df.index == P * period, "◀", "")
        result_df["Q"] = np.where(result_df.index == Q * period, "◀", "")

    # --- 8) 판정 결과 출력 ---
    print(f"관측치 {n}개 | 유의성 기준값 {threshold:.3f} | "
            f"최대 시차 {lags} (상한 N/4={limit})")

    for name, code, order, kind, step in (("AR", "p", p, "PACF", 1), ("MA", "q", q, "ACF", 1)):
        reason = (f"{kind}가 lag {order * step} 이후 절단" if order
                    else f"{kind}가 lag {step}부터 비유의")
        print(f"{name} 차수 후보 ({code}): {order}   ← {reason}")

    if period is not None:
        for name, code, order, kind, step in (("계절 AR", "P", P, "계절 PACF", period),
                                                ("계절 MA", "Q", Q, "계절 ACF", period)):
            reason = (f"{kind}가 lag {order * step} 이후 절단" if order
                        else f"{kind}가 lag {step}부터 비유의")
            print(f"{name} 차수 후보 ({code}): {order}   ← {reason} "
                    f"(판정 시차 {season_lags})")

        print(f"→ SARIMA({p}, d, {q})({P}, D, {Q})[{period}] "
                f"— d·D 는 차분 단계의 결과를 넣는다")

    # --- 9) 시각화 ---
    if plot:
        fig, ax = my_plot.init(rows=rows, cols=cols, width=width, height=height, title=title)

        # ACF : MA 차수(q·Q)의 근거 / PACF : AR 차수(p·P)의 근거
        for axis, func, name, code, order, s_code, s_order in (
                (ax[0], plot_acf, "ACF", "q", q, "Q", Q),
                (ax[1], plot_pacf, "PACF", "p", p, "P", P)):
            func(data, lags=lags, ax=axis)

            if order:
                axis.axvline(order, color="red", linestyle="--", linewidth=1)
                subtitle = f"{name} — lag {order} 이후 절단 → {code} = {order}"
            else:
                subtitle = f"{name} — lag 1부터 비유의 → {code} = 0"

            if period is not None and s_order:
                # 계절 차수가 확정된 시차는 초록 점선으로 따로 표시한다
                axis.axvline(s_order * period, color="green", linestyle=":", linewidth=1.5)
                subtitle += f" | 계절 lag {s_order * period}까지 유의 → {s_code} = {s_order}"

            axis.set_title(subtitle)
            axis.set_xlabel("Lag")

        my_plot.show(save_path=save_path)

    # --- 10) 가장 뒤쪽 차수의 lag까지만 반환 ---
    # 차수가 하나도 잡히지 않았다면 자를 기준이 없으므로 근거를 통째로 돌려준다
    cut = max(p, q, P * period, Q * period) if period else max(p, q)

    return result_df if cut == 0 else result_df.loc[:cut]

fit_model

fit_model(
    data,
    column=None,
    p=1,
    d=1,
    q=1,
    P=1,
    D=1,
    Q=1,
    s=None,
    log=False,
    search_diff=False,
    report=True,
    summary=False,
)

차수 범위 안의 모든 SARIMA 조합을 적합해 AIC가 가장 낮은 모형을 고른다.

Parameters:

Name	Type	Description	Default
data	DataFrame | Series	차분하지 않은 원본 시계열.	required
column	str	data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).	None
p	int	비계절 AR 차수의 상한 (기본값: 1).	1
d	int	일반 차분 횟수 (기본값: 1).	1
q	int	비계절 MA 차수의 상한 (기본값: 1).	1
P	int	계절 AR 차수의 상한 (기본값: 1).	1
D	int	계절 차분 횟수 (기본값: 1).	1
Q	int	계절 MA 차수의 상한 (기본값: 1).	1
s	int	계절 주기. None이면 계절 성분을 쓰지 않음 (기본값: None).	None
log	bool	로그를 씌워 적합할지 여부 (기본값: False).	False
search_diff	bool	d·D도 0부터 탐색할지 여부 (기본값: False).	False
report	bool	조합별 AIC·BIC 결과표 출력 여부 (기본값: True).	True
summary	bool	선택된 모형의 요약 통계량 출력 여부 (기본값: False).	False

Returns:

Type	Description
	선택된 차수로 적합한 SARIMAX 결과 객체. 탐색 결과표가 grid_, 로그변환 여부가 log_ 속성으로 붙는다.

Source code in jussam/my_ts.py

def fit_model(data, column=None, p=1, d=1, q=1, P=1, D=1, Q=1, s=None,
              log=False, search_diff=False, report=True, summary=False):
    """차수 범위 안의 모든 SARIMA 조합을 적합해 AIC가 가장 낮은 모형을 고른다.

    Args:
        data (DataFrame | Series): 차분하지 않은 원본 시계열.
        column (str): data가 데이터프레임인 경우 대상 컬럼명 (기본값: None).
        p (int): 비계절 AR 차수의 상한 (기본값: 1).
        d (int): 일반 차분 횟수 (기본값: 1).
        q (int): 비계절 MA 차수의 상한 (기본값: 1).
        P (int): 계절 AR 차수의 상한 (기본값: 1).
        D (int): 계절 차분 횟수 (기본값: 1).
        Q (int): 계절 MA 차수의 상한 (기본값: 1).
        s (int): 계절 주기. None이면 계절 성분을 쓰지 않음 (기본값: None).
        log (bool): 로그를 씌워 적합할지 여부 (기본값: False).
        search_diff (bool): d·D도 0부터 탐색할지 여부 (기본값: False).
        report (bool): 조합별 AIC·BIC 결과표 출력 여부 (기본값: True).
        summary (bool): 선택된 모형의 요약 통계량 출력 여부 (기본값: False).

    Returns:
        선택된 차수로 적합한 SARIMAX 결과 객체.
            탐색 결과표가 `grid_`, 로그변환 여부가 `log_` 속성으로 붙는다.
    """
    # --- 1) 대상 추출 ---
    if column is not None:
        data = data[column]

    data = data.dropna()

    # 로그변환은 여기서 한 번만 한다. 차분은 SARIMAX가 d·D로 알아서 처리한다
    endog = np.log(data) if log else data

    # --- 2) 탐색할 차수 조합 만들기 ---
    # 계절 주기가 없으면 계절 성분은 아예 만들지 않는다
    if s is None:
        s = P = D = Q = 0

    # d·D는 차분 단계에서 이미 확정된 값이므로 기본적으로 고정한다.
    # 차분 횟수가 다르면 모형이 설명하는 대상 자체가 달라져 AIC를 나란히 비교할 수 없다.
    d_range = range(d + 1) if search_diff else [d]
    D_range = range(D + 1) if search_diff else [D]

    orders = list(product(range(p + 1), d_range, range(q + 1),
                          range(P + 1), D_range, range(Q + 1)))

    # --- 3) 조합마다 적합해 AIC·BIC를 쌓는다 ---
    result = []     # 결과를 담을 리스트
    best = None     # 최적 모형을 담을 변수

    for try_p, try_d, try_q, try_P, try_D, try_Q in orders:
        # --- 3-1) 적합 시도 ---
        try:
            # 조합을 여러 개 적합하면 수렴 경고가 반복 출력되므로 이 안에서만 숨긴다
            with warnings.catch_warnings():
                warnings.simplefilter("ignore")
                m = SARIMAX(endog, order=(try_p, try_d, try_q),
                            seasonal_order=(try_P, try_D, try_Q, s)).fit(disp=False)
        except Exception:
            continue                # 적합에 실패한 조합은 후보에서 뺀다

        # --- 3-2) 결과 기록 ---
        name = f"({try_p},{try_d},{try_q})"

        if s:
            name += f"({try_P},{try_D},{try_Q},{s})"

        result.append({
            "모형": name,
            "p": try_p, "d": try_d, "q": try_q,
            "P": try_P, "D": try_D, "Q": try_Q, "s": s,
            "계수 수": len(m.params),
            "AIC": round(m.aic, 3),
            "BIC": round(m.bic, 3),
        })

        # --- 3-3) 최적 모형 갱신 ---
        # AIC가 더 낮은 모형을 만나면 그 자리에서 갈아 끼운다 (다시 적합할 필요가 없다)
        if best is None or m.aic < best.aic:
            best = m

    if best is None:
        raise ValueError("적합에 성공한 조합이 없습니다. 차수 범위나 데이터를 확인하세요.")

    # --- 4) AIC 오름차순으로 순위를 매긴다 ---
    grid = DataFrame(result).sort_values("AIC").reset_index(drop=True)
    grid.index = grid.index + 1
    grid.index.name = "순위"

    # --- 5) 탐색 결과 출력 ---
    if report:
        print(f"시도한 조합 {len(orders)}가지 | 적합 성공 {len(grid)}가지"
              f"{' | 로그변환 적용' if log else ''}")
        print(f"AIC 최적: {grid.iloc[0]['모형']}  |  "
              f"BIC 최적: {grid.sort_values('BIC').iloc[0]['모형']}")
        display(grid)

    if summary:
        print(best.summary())

    # --- 6) 탐색 결과와 로그 여부를 결과 객체에 붙여 반환 ---
    # predict() 가 예측값을 원래 단위로 되돌릴 때 log_ 를 쓴다
    best.grid_ = grid
    best.log_ = log

    return best

predict

predict(
    fit,
    steps,
    alpha=0.05,
    plot=True,
    history=None,
    title=None,
    xlabel=None,
    ylabel=None,
    width=1280,
    height=520,
    save_path=None,
)

적합된 SARIMA 모형으로 관측 구간 다음의 예측값과 예측구간을 계산한다.

Parameters:

Name	Type	Description	Default
fit		fit_model 함수로 적합된 SARIMAX 결과 객체.	required
steps	int	내다볼 시점의 개수. 계절 주기를 넣으면 한 시즌이 된다.	required
alpha	float	예측구간의 유의수준. 0.05면 95% 구간 (기본값: 0.05).	0.05
plot	bool	관측값과 예측 결과를 그래프로 그릴지 여부 (기본값: True).	True
history	int	그래프에 함께 그릴 최근 관측치 수. None이면 전체 (기본값: None).	None
title	str	그래프 제목 (기본값: None).	None
xlabel	str	x축 레이블 (기본값: None).	None
ylabel	str	y축 레이블 (기본값: None).	None
width	int	캔버스 가로 픽셀 (기본값: 1280).	1280
height	int	캔버스 세로 픽셀 (기본값: 520).	520
save_path	str	이미지 저장 경로 (기본값: None).	None

Returns:

Name	Type	Description
DataFrame		미래 시점을 인덱스로 하는 예측 표 (pred · lower · upper).

Source code in jussam/my_ts.py

def predict(fit, steps, alpha=0.05, plot=True, history=None,
            title=None, xlabel=None, ylabel=None,
            width=1280, height=520, save_path=None):
    """적합된 SARIMA 모형으로 관측 구간 다음의 예측값과 예측구간을 계산한다.

    Args:
        fit: `fit_model` 함수로 적합된 SARIMAX 결과 객체.
        steps (int): 내다볼 시점의 개수. 계절 주기를 넣으면 한 시즌이 된다.
        alpha (float): 예측구간의 유의수준. 0.05면 95% 구간 (기본값: 0.05).
        plot (bool): 관측값과 예측 결과를 그래프로 그릴지 여부 (기본값: True).
        history (int): 그래프에 함께 그릴 최근 관측치 수. None이면 전체 (기본값: None).
        title (str): 그래프 제목 (기본값: None).
        xlabel (str): x축 레이블 (기본값: None).
        ylabel (str): y축 레이블 (기본값: None).
        width (int): 캔버스 가로 픽셀 (기본값: 1280).
        height (int): 캔버스 세로 픽셀 (기본값: 520).
        save_path (str): 이미지 저장 경로 (기본값: None).

    Returns:
        DataFrame: 미래 시점을 인덱스로 하는 예측 표 (pred · lower · upper).
    """
    # --- 1) 예측값과 예측구간 계산 ---
    # 평균 예측값과 예측구간을 함께 담고 있는 결과 객체
    forecast = fit.get_forecast(steps=steps)
    conf = forecast.conf_int(alpha=alpha)

    result = DataFrame({
        "pred": forecast.predicted_mean,
        "lower": conf.iloc[:, 0],
        "upper": conf.iloc[:, 1],
    })

    # 로그를 씌워 적합했다면 예측값도 로그 스케일이므로 원래 단위로 되돌린다
    log = getattr(fit, "log_", False)

    if log:
        result = np.exp(result)

    result.index.name = "시점"  # 인덱스 이름 설정

    if not plot:               # 그래프 옵션 off시 결과 반환 및 함수 종료
        return result

    # --- 2) 그래프에 함께 그릴 관측 구간 준비 ---
    # 적합에 쓴 원본은 결과 객체가 그대로 들고 있다. 따로 넘겨받을 필요가 없다
    observed = Series(np.asarray(fit.model.endog).ravel(),
                      index=fit.fittedvalues.index)

    if log:
        observed = np.exp(observed)

    if history is not None:         # 예측 구간 근처만 확대해서 보고 싶을 때
        observed = observed.iloc[-history:]

    # --- 3) 그래프 제목 및 데이터 연결  ---
    # 그래프 제목 구성
    if title is None:
        title = f"{steps}시점 예측 ({(1 - alpha) * 100:.0f}% 예측구간)"

    # 적합값과 예측치를 연결
    # -> 관측의 마지막 값을 예측선의 출발점으로 붙여야 선이 끊겨 보이지 않는다
    last = observed.index[-1]
    linked = concat([Series({last: observed.iloc[-1]}), result["pred"]])

    # --- 4) 그래프 그리기 ---
    fig, ax = my_plot.init(width=width, height=height, title=title,
                           xlabel=xlabel, ylabel=ylabel)

    my_plot.lineplot(x=observed.index, y=observed.values, label="관측값", ax=ax)

    my_plot.lineplot(x=linked.index, y=linked.values, label="예측값",
                     linestyle="--", ax=ax)

    # 예측구간은 직선이 아니라 점선으로 그린다. 뒤로 갈수록 넓어지는 모습이 드러난다
    ax.fill_between(result.index, result["lower"], result["upper"],
                    alpha=0.2, label=f"{(1 - alpha) * 100:.0f}% 예측구간")

    # 경계 표시: 이 선의 오른쪽은 실제 값이 존재하지 않는 구간이다
    ax.axvline(last, color="gray", linestyle=":", linewidth=1)

    ax.legend()
    my_plot.show(save_path=save_path)

    return result

report_residual

report_residual(
    fit,
    alpha=0.05,
    report=True,
    plot=True,
    title=None,
    xlabel=None,
    ylabel=None,
    width=800,
    height=480,
    save_path=None,
)

적합된 시계열 모형의 잔차가 세 가지 가정을 만족하는지 한 표로 점검한다.

잔차에 패턴이 남아 있으면 모형이 아직 뽑아내지 못한 정보가 있다는 뜻이고, 정규성·등분산성이 깨지면 예측구간의 폭을 그대로 믿을 수 없게 된다.

Parameters:

Name	Type	Description	Default
fit		적합된 시계열 모형 결과 객체.	required
alpha	float	유의수준 (기본값: 0.05).	0.05
report	bool	종합 판정 문장 출력 여부 (기본값: True).	True
plot	bool	잔차의 흐름과 분포를 그래프로 그릴지 여부 (기본값: True).	True
title	str	그래프 전체 제목 (기본값: None).	None
xlabel	str	잔차 흐름 그래프의 x축 레이블 (기본값: None).	None
ylabel	str	잔차 흐름 그래프의 y축 레이블 (기본값: None).	None
width	int	캔버스 한 칸의 가로 픽셀 (기본값: 800).	800
height	int	캔버스 세로 픽셀 (기본값: 480).	480
save_path	str	이미지 저장 경로 (기본값: None).	None

Returns:

Name	Type	Description
DataFrame		검정명을 인덱스로 하는 잔차 진단 결과표.

Source code in jussam/my_ts.py

def report_residual(fit, alpha=0.05, report=True, plot=True, title=None, 
                    xlabel=None, ylabel=None, width=800, height=480, save_path=None):
    """적합된 시계열 모형의 잔차가 세 가지 가정을 만족하는지 한 표로 점검한다.

    잔차에 패턴이 남아 있으면 모형이 아직 뽑아내지 못한 정보가 있다는 뜻이고,
    정규성·등분산성이 깨지면 예측구간의 폭을 그대로 믿을 수 없게 된다.

    Args:
        fit: 적합된 시계열 모형 결과 객체.
        alpha (float): 유의수준 (기본값: 0.05).
        report (bool): 종합 판정 문장 출력 여부 (기본값: True).
        plot (bool): 잔차의 흐름과 분포를 그래프로 그릴지 여부 (기본값: True).
        title (str): 그래프 전체 제목 (기본값: None).
        xlabel (str): 잔차 흐름 그래프의 x축 레이블 (기본값: None).
        ylabel (str): 잔차 흐름 그래프의 y축 레이블 (기본값: None).
        width (int): 캔버스 한 칸의 가로 픽셀 (기본값: 800).
        height (int): 캔버스 세로 픽셀 (기본값: 480).
        save_path (str): 이미지 저장 경로 (기본값: None).

    Returns:
        DataFrame: 검정명을 인덱스로 하는 잔차 진단 결과표.
    """
    # --- 1) 모형에서 차수 정보를 꺼낸다 ---
    # SARIMAX 계열은 (p,d,q)와 (P,D,Q,s)를 모형 객체가 그대로 들고 있다.
    # 여기서 쓰는 값은 차분 횟수(d·D)와 계절 주기(period)뿐이다
    model = getattr(fit, "model", None)
    p, d, q = getattr(model, "order", (0, 0, 0))
    P, D, Q, period = getattr(model, "seasonal_order", (0, 0, 0, 0))

    # --- 2) 진단 대상 잔차 준비 ---
    x = Series(fit.resid).dropna()

    # 차분이 소모한 앞 구간은 모형이 아직 자리를 잡지 못해 잔차가 비정상적으로 크다
    burn = d + D * period

    if burn:
        x = x.iloc[burn:]

    result = []     # 검정 결과를 담을 리스트

    # --- 3) 무상관 : Ljung-Box 검정 ---
    # 여러 시차를 한 묶음으로 검정해 "남은 자기상관이 있는가"를 판정한다
    # 계절 주기가 있으면 한 시즌·두 시즌, 없으면 관례값인 10 시차를 본다
    # (Hyndman & Athanasopoulos, 2021), statsmodels 내부 코드도 10 시차를 기본값으로 쓴다
    lags = [period, period * 2] if period else [10]

    # 모형이 추정한 계수 수만큼 자유도를 빼야 한다. 시차가 이보다 작으면 검정이 성립하지 않는다
    model_df = p + q + P + Q

    # 시차가 표본 대비 크면 검정력이 떨어진다. 관례에 따라 관측치 수의 1/5로 제한한다
    lags = [lag for lag in lags if model_df < lag <= len(x) // 5]

    # 시차가 하나도 남지 않으면 관례값을 쓰되, 관측치 수의 1/5보다 크면 안 된다
    if not lags: lags = [max(model_df + 1, len(x) // 5)]

    # 무상관 검정 수행
    lb = acorr_ljungbox(x, lags=lags, model_df=model_df, return_df=True)

    # 검정 결과를 표로 정리
    for lag, row in lb.iterrows():
        passed = bool(row["lb_pvalue"] >= alpha)

        result.append({
            "검정": f"Ljung-Box (시차 {lag})",
            "가정": "무상관",
            "관측치 수": len(x),
            "통계량": round(row["lb_stat"], 3),
            "p-value": round(row["lb_pvalue"], 4),
            "통과": passed,
            "판정": "통과 (자기상관 없음)" if passed else "실패 (자기상관 남음)",
            "비고": f"자유도 {lag - model_df}",
        })

    # --- 4) 정규성 : Jarque-Bera 검정 ---
    jb_stat, jb_p, skew, kurtosis = fit.test_normality("jarquebera")[0]

    passed = bool(jb_p >= alpha)

    result.append({
        "검정": "Jarque-Bera",
        "가정": "정규성",
        "관측치 수": len(x),
        "통계량": round(jb_stat, 3),
        "p-value": round(jb_p, 4),
        "통과": passed,
        "판정": "통과 (정규성 만족)" if passed else "실패 (정규성 위배)",
        "비고": f"왜도 {skew:.3f} · 첨도 {kurtosis:.3f}",
    })

    # --- 5) 등분산성 : 이분산(H) 검정 ---
    h_stat, h_p = fit.test_heteroskedasticity("breakvar")[0]

    passed = bool(h_p >= alpha)     # 이분산 검정 통과 여부

    # 검정 결과를 표로 정리
    result.append({
        "검정": "이분산(H)",
        "가정": "등분산성",
        "관측치 수": len(x),
        "통계량": round(h_stat, 3),
        "p-value": round(h_p, 4),
        "통과": passed,
        "판정": "통과 (등분산)" if passed else "실패 (이분산)",
        # H > 1이면 예측구간이 좁게(낙관적), H < 1이면 넓게(보수적) 잡힌다
        "비고": "뒤로 갈수록 커짐" if h_stat > 1 else "뒤로 갈수록 작아짐",
    })

    # --- 6) 결과표 구성 ---
    result = DataFrame(result).set_index("검정")

    if report:
        failed = list(result.index[~result["통과"]])

        print(f"진단 대상 {len(x)}개 (앞 {burn}개 제외) | 유의수준 {alpha}")
        print("실패한 검정:", ", ".join(failed) if failed else "없음 (모든 가정 통과)")

    if not plot:               # 그래프 옵션 off시 결과 반환 및 함수 종료
        return result

    # --- 7) 잔차의 모습 살펴보기 ---
    # 검정 결과는 숫자 하나로만 말해 준다. 눈으로 봐야 잡히는 문제도 있다
    fig, ax = my_plot.init(rows=1, cols=2, width=width, height=height, title=title)

    # 시간에 따른 잔차 : 0 주변을 무작위로 오가야 한다
    my_plot.lineplot(x=x.index, y=x.values, ax=ax[0])
    ax[0].axhline(0, color="red", linestyle="--", linewidth=1)
    ax[0].set_title("시간에 따른 잔차")
    ax[0].set_xlabel(xlabel)
    ax[0].set_ylabel(ylabel)

    # 잔차의 분포 : 종 모양이어야 한다
    my_plot.histplot(data=x.to_frame(name="잔차"), x="잔차", kde=True, ax=ax[1])
    ax[1].set_title("잔차의 분포")

    my_plot.show(save_path=save_path)

    return result

report_score

report_score(fit, name='적합값', report=True)

모형의 예측값과 기준선(나이브)의 오차를 원래 단위에서 나란히 비교한다.

성능 지표는 혼자 보면 잘한 것인지 알 수 없다. "작년 같은 달과 똑같을 것이다" 수준의 기준선을 함께 계산해 비교 대상으로 삼는다.

Parameters:

Name	Type	Description	Default
fit		적합된 시계열 모형 결과 객체.	required
name	str	결과표에 표시할 모형 행 이름 (기본값: "적합값").	'적합값'
report	bool	평가 구간과 기준선 대비 개선폭 출력 여부 (기본값: True).	True

Returns:

Name	Type	Description
DataFrame		대상을 인덱스로 하는 성능 지표표 (MAE · RMSE · MAPE).

Source code in jussam/my_ts.py

def report_score(fit, name="적합값", report=True):
    """모형의 예측값과 기준선(나이브)의 오차를 원래 단위에서 나란히 비교한다.

    성능 지표는 혼자 보면 잘한 것인지 알 수 없다.
    "작년 같은 달과 똑같을 것이다" 수준의 기준선을 함께 계산해 비교 대상으로 삼는다.

    Args:
        fit: 적합된 시계열 모형 결과 객체.
        name (str): 결과표에 표시할 모형 행 이름 (기본값: "적합값").
        report (bool): 평가 구간과 기준선 대비 개선폭 출력 여부 (기본값: True).

    Returns:
        DataFrame: 대상을 인덱스로 하는 성능 지표표 (MAE · RMSE · MAPE).
    """
    # --- 1) 모형에서 차수 정보를 꺼낸다 ---
    # 여기서 쓰는 값은 차분 횟수(d·D)와 계절 주기(s)뿐이다
    model = getattr(fit, "model", None)
    p, d, q = getattr(model, "order", (0, 0, 0))
    P, D, Q, s = getattr(model, "seasonal_order", (0, 0, 0, 0))

    # 로그를 씌워 적합했다면 지표는 반드시 원래 단위로 되돌린 뒤 계산해야
    # "몇 명 틀리는가"를 말할 수 있다
    log = getattr(fit, "log_", False)

    # --- 2) 관측값과 예측값 확보 ---
    # 적합에 쓴 시계열은 결과 객체가 그대로 들고 있다. 따로 넘겨받을 필요가 없다
    endog = model.data.orig_endog

    actual = endog if isinstance(endog, Series) else \
        Series(np.asarray(endog).ravel(), index=fit.fittedvalues.index)

    pred = fit.fittedvalues

    if log:
        actual = np.exp(actual)
        pred = np.exp(pred)

    # --- 3) 기준선(나이브) 준비 ---
    # 계절 주기가 있으면 작년 같은 달, 없으면 직전 시점을 그대로 예측값으로 쓴다
    period = s if s else 1

    base_name = f"계절 나이브 (시차 {period})" if period > 1 else "나이브 (직전값)"

    # --- 4) 공통 평가 구간 맞추기 ---
    # 대상마다 유효 구간이 다르면 비교가 성립하지 않는다. 모두 존재하는 구간만 남긴다
    table = DataFrame({
        "관측값": actual,
        name: pred,
        base_name: actual.shift(period),    # 기준선은 항상 함께 계산한다
    })

    # 차분이 소모한 앞 구간은 모형이 아직 자리를 잡지 못해 오차가 비정상적으로 크다
    burn = d + D * s

    # 차분이 소모한 구간이 있다면 그 구간은 평가에서 제외
    if burn: table = table.iloc[burn:]

    # 관측값이 없는 시점은 평가에서 제외
    table = table.dropna()

    # --- 5) 대상별 지표 계산 ---
    y = table["관측값"].values   # 관측값
    nonzero = y != 0            # 관측값이 0인 시점은 MAPE를 계산할 수 없다
    result = []                 # 지표를 담을 리스트

    for target in table.columns[1:]:        # 0번째 컬럼은 관측값이므로 제외
        error = y - table[target].values    # 오차 계산 (관측값 − 예측값)

        result.append({
            "대상": target,
            "관측치 수": len(error),
            "MAE": round(np.mean(np.abs(error)), 3),
            "RMSE": round(np.sqrt(np.mean(error ** 2)), 3),
            "MAPE(%)": round(np.mean(np.abs(error[nonzero] / y[nonzero])) * 100, 3)
            if nonzero.any() else np.nan,
        })

    result = DataFrame(result).set_index("대상")

    # --- 6) 기준선 대비 개선폭 출력 및 반환 ---
    if report:
        model_mape = result.loc[name, "MAPE(%)"]
        base_mape = result.loc[base_name, "MAPE(%)"]
        gap = (base_mape - model_mape) / base_mape * 100

        print(f"평가 구간 {len(table)}개 ({table.index[0]} ~ {table.index[-1]})")
        print(f"MAPE {base_mape:.3f}% → {model_mape:.3f}% | "
              f"기준선 대비 {abs(gap):.1f}% {'개선' if gap > 0 else '악화'}")

    return result

granger_test

granger_test(
    data, y, x=None, maxlag=4, alpha=0.05, report=True
)

독립변수들이 종속변수를 그랜저 인과하는지 시차별로 반복 검정한다.

그랜저 인과는 "원인"이 아니라 "예측에 도움이 되는가"를 본다. x의 과거값을 함께 넣었을 때 y의 예측 오차가 유의하게 줄어들면 인과가 있다고 본다.

검정 대상은 반드시 정상 시계열이어야 한다. 비정상 시계열을 그대로 넣으면 추세끼리 닮았다는 이유만으로 인과가 있다고 나온다(허구적 인과). 그래서 검정 전에 ADF 검정으로 정상성을 먼저 점검하고 경고한다.

Parameters:

Name	Type	Description	Default
data	DataFrame	종속변수와 독립변수가 모두 들어있는 시계열 데이터프레임.	required
y	str	종속변수(결과) 컬럼명.	required
x	str | list	독립변수(원인 후보) 컬럼명 또는 컬럼명 리스트 (기본값: None). 생략하면 y를 제외한 나머지 컬럼 전부를 독립변수로 사용한다.	None
maxlag	int	검정할 최대 시차 (기본값: 4).	4
alpha	float	유의수준 (기본값: 0.05).	0.05
report	bool	정상성 경고와 변수별 요약 출력 여부 (기본값: True).	True

Returns:

Name	Type	Description
DataFrame		독립변수 · 시차를 인덱스로 하는 검정 결과표. "권장" 컬럼에 ★ 표시가 붙은 시차가 그 변수를 shift() 할 시차다.

Source code in jussam/my_ts.py

def granger_test(data, y, x=None, maxlag=4, alpha=0.05, report=True):
    """독립변수들이 종속변수를 그랜저 인과하는지 시차별로 반복 검정한다.

    그랜저 인과는 "원인"이 아니라 "예측에 도움이 되는가"를 본다.
    x의 과거값을 함께 넣었을 때 y의 예측 오차가 유의하게 줄어들면 인과가 있다고 본다.

    검정 대상은 반드시 정상 시계열이어야 한다. 비정상 시계열을 그대로 넣으면
    추세끼리 닮았다는 이유만으로 인과가 있다고 나온다(허구적 인과).
    그래서 검정 전에 ADF 검정으로 정상성을 먼저 점검하고 경고한다.

    Args:
        data (DataFrame): 종속변수와 독립변수가 모두 들어있는 시계열 데이터프레임.
        y (str): 종속변수(결과) 컬럼명.
        x (str | list): 독립변수(원인 후보) 컬럼명 또는 컬럼명 리스트 (기본값: None).
            생략하면 y를 제외한 나머지 컬럼 전부를 독립변수로 사용한다.
        maxlag (int): 검정할 최대 시차 (기본값: 4).
        alpha (float): 유의수준 (기본값: 0.05).
        report (bool): 정상성 경고와 변수별 요약 출력 여부 (기본값: True).

    Returns:
        DataFrame: 독립변수 · 시차를 인덱스로 하는 검정 결과표.
            "권장" 컬럼에 ★ 표시가 붙은 시차가 그 변수를 shift() 할 시차다.
    """
    # --- 0) 검정 대상 정리 ---
    if x is None:                                   # 독립변수를 지정하지 않았다면
        x = [c for c in data.columns if c != y]     # y를 뺀 나머지 전부를 대상으로
    elif isinstance(x, str):                        # 컬럼명 하나를 문자열로 준 경우
        x = [x]                                     # 리스트로 감싸 아래 반복문에 맞춘다

    # --- 1) 정상성 사전 점검 : 허구적 인과를 막는다 ---
    if report:
        checked = concat([adf_test(data, column=c, alpha=alpha) for c in [y] + x])
        unstable = list(checked[~checked["정상성"]].index)   # 비정상으로 판정된 컬럼들

        if unstable:
            print(f"⚠️ 비정상 시계열이 포함되어 있습니다: {', '.join(unstable)}")
            print("   차분·로그변환으로 정상성을 확보한 뒤 검정하세요. → adf_transform()")

    result = []             # 검정 결과를 담을 리스트

    # --- 2) 독립변수마다 반복 검정 ---
    for name in x:
        # 그랜저 인과 검정은 [결과, 원인] 순서로 두 컬럼을 넘긴다
        pair = data[[y, name]].dropna()

        # 검정 보고서가 변수마다 쏟아지므로 어디서 변수가 바뀌는지 구분선을 찍어 둔다
        if report:
            print(f"\n{'=' * 70}")
            print(f"■ {name} → {y}")
            print(f"{'=' * 70}")

        # 시차마다 회귀식을 세우므로 관측치가 부족하면 검정 자체가 성립하지 않는다
        try:
            # 시차마다 검정 결과가 화면에 출력되고, 반환값으로도 딕셔너리를 받는다
            tests = grangercausalitytests(pair, maxlag=maxlag)
        except Exception as e:
            if report:  print(f"⚠️ {name}: 검정을 수행할 수 없습니다. ({e})")
            continue

        # --- 2-1) 시차별 F검정 결과를 한 행씩 쌓는다 ---
        for lag, test in tests.items():
            # ssr_ftest = 잔차제곱합 기반 F검정. (F통계량, p-value, 자유도, 시차)
            statistic, pvalue, dfd, dfn = test[0]["ssr_ftest"]
            causal = bool(pvalue < alpha)       # p-value가 작아야 인과 있음

            result.append({
                "독립변수": name,
                "시차": lag,
                "관측치 수": len(pair),
                "F 통계량": round(statistic, 3),
                "p-value": round(pvalue, 4),
                "인과": causal,
                "판정": "인과 있음" if causal else "인과 없음",
            })

    if not result:                              # 검정에 성공한 변수가 하나도 없다면
        return DataFrame()                      # 빈 결과표를 반환

    result_df = DataFrame(result)

    # --- 3) 변수별 권장 시차 표시 ---
    # 유의한 시차가 여럿이면 p-value가 가장 작은 시차가 신호가 가장 뚜렷한 지점이다
    recommend = []                              # 권장 표시를 담을 리스트
    best_lag = {}                               # 변수별 권장 시차 (아래 요약 출력용)

    for name in result_df["독립변수"].unique():
        block = result_df[result_df["독립변수"] == name]     # 그 변수의 시차별 결과
        causal = block[block["인과"]]                        # 그 중 유의한 시차만

        # 유의한 시차가 없으면 권장 시차도 없다
        best_lag[name] = causal["p-value"].idxmin() if len(causal) else None

    for i in result_df.index:                               # 각 행을 순서대로 순회하며
        name = result_df.loc[i, "독립변수"]
        recommend.append("★ 권장 시차" if i == best_lag[name] else "")

    result_df["권장"] = recommend               # 권장 표시를 결과표에 추가

    # --- 4) 변수별 요약 출력 ---
    if report:
        print(f"\n종속변수 {y} ← 독립변수 {len(best_lag)}개 · 최대 시차 {maxlag}")

        for name, i in best_lag.items():
            if i is None:                       # 유의한 시차가 하나도 없었다면
                print(f"  {name} → {y} : 인과 없음 (시차 1~{maxlag} 모두 p ≥ {alpha})")
            else:
                lag = result_df.loc[i, "시차"]
                pvalue = result_df.loc[i, "p-value"]
                print(f"  {name} → {y} : 시차 {lag}에서 인과 있음 (p-value {pvalue:.4f}) "
                      f"→ {name}.shift({lag})")

    return result_df.set_index(["독립변수", "시차"])