Python 사용자 지정 월별 집계

Python 함수는 온톨로지 집계 결과를 가공한 뒤 Workshop 차트가 읽는 ThreeDimensionalAggregation으로 반환할 수 있습니다. 이 예제는 부서별 최근 관측 월의 총지출을 다음 여섯 달에 반복합니다.

TwoDimensionalAggregation.from_osdk와 ThreeDimensionalAggregation.from_osdk는 Python OSDK v2에서 지원합니다. 이 함수에도 v2 생성 SDK가 필요합니다.

입력과 날짜별 합계

가상의 Expense에 String department_name, Date date와 Double amount가 있다고 가정합니다. 온톨로지 API 이름은 각각 departmentName, date와 amount입니다.

OSDK에서 department_name.exact()와 date.exact()로 그룹을 나누고 amount.sum을 계산합니다. date.exact는 날짜별 그룹이므로 같은 달의 여러 날짜를 월 합계로 다시 합쳐야 합니다. from_osdk에는 Python 속성 이름 대신 API 이름 departmentName과 date를 전달합니다.

최신 월과 여섯 개 구간

날짜를 월 첫날로 정규화하여 부서별 월 합계를 만듭니다. 월 키의 최댓값으로 마지막 관측 월을 찾고 그 다음 달부터 여섯 개의 반열린 구간 [min, max)을 생성합니다. 달의 길이나 연도 경계를 30일 고정 값으로 계산하지 않습니다.

from datetime import date
from functions.api import (
    Date, Double, NestedBucket, SingleBucket, Range,
    ThreeDimensionalAggregation, UserFacingError, function,
)
from ontology_sdk import FoundryClient
from ontology_sdk.ontology.objects import Expense


def next_month(month: date) -> date:
    if month.month == 12:
        return date(month.year + 1, 1, 1)
    return date(month.year, month.month + 1, 1)


@function
def estimated_department_expenses() -> ThreeDimensionalAggregation[str, Range[Date], Double]:
    client = FoundryClient()
    result = (
        client.ontology.objects.Expense
        .group_by(Expense.object_type.department_name.exact())
        .group_by(Expense.object_type.date.exact())
        .sum(Expense.object_type.amount)
        .compute()
    )
    current = ThreeDimensionalAggregation.from_osdk(result, "departmentName", "date")
    forecasts: list[NestedBucket[str, Range[Date], Double]] = []

    for department in current.buckets:
        monthly: dict[date, float] = {}
        for daily in department.buckets:
            if daily.key is None or daily.value is None:
                raise UserFacingError("지출 집계에는 날짜와 합계가 필요합니다.")
            month = date.fromisoformat(str(daily.key)).replace(day=1)
            monthly[month] = monthly.get(month, 0.0) + daily.value

        buckets: list[SingleBucket[Range[Date], Double]] = []
        if monthly:
            latest_month = max(monthly)
            baseline = monthly[latest_month]
            start = next_month(latest_month)
            for _ in range(6):
                end = next_month(start)
                buckets.append(SingleBucket(Range(min=start, max=end), baseline))
                start = end
        forecasts.append(NestedBucket(department.key, buckets))

    return ThreeDimensionalAggregation(forecasts)

각 날짜의 그룹 키는 YYYY-MM-DD로 읽습니다. 반환할 때는 실제 Date 값을 Range에 넣고, NestedBucket이 부서를, SingleBucket이 월별 값 하나를 나타냅니다. 빈 부서 데이터에는 전망 버킷이 없으며 누락 날짜나 합계에서 임의의 예측 값을 만들지 않습니다.

from_osdk는 집계 결과의 구조를 변환합니다. 원격 그룹 결과가 누락되거나 근사되었다면 변환만으로 전체 월합계를 복원할 수 없으므로 원격 집계의 범위와 완전성을 확인해야 합니다.

전망값 해석

마지막 관측 월에 일부 날짜만 들어 있다면 이 함수는 그 부분 합계도 그대로 다음 달에 반복합니다. 기준 월의 완전성, 추세와 계절성을 보정하지 않는 단순 전망입니다.

게시한 함수의 반환 타입을 Workshop 차트에 연결할 수 있습니다. TSv1 예제는 서버에서 byMonth로 그룹화하고 Timestamp 구간을 바로 받는 방식입니다. 두 언어의 내부 버킷 구조가 달라도 차트에는 같은 부서별 월 전망을 제공합니다.