Create a custom aggregation（创建自定义聚合）

Functions can be used to compute custom aggregations based on data in the ontology, which can then be surfaced in a chart widget in Workshop. This guide walks through how to write custom aggregation logic that loads aggregated data from the ontology, manipulates the results to create a projection of future results, and returns the modified results.

These references may be useful while working through this section:

• Reference for Python aggregation types

:::callout{theme="warning"} .from_osdk() in TwoDimensionalAggregation and ThreeDimensionalAggregation are only supported when using v2 of the Python OSDK. :::

Loading an aggregation

In this example, assume you have an ontology consisting of expenses, with each expense object having properties for department name, expense date, and expense amount. If you want to estimate the monthly spend by department over the next six months, you can begin by loading the aggregated data for the monthly spend:

client = FoundryClient()
result: AggregateObjectsResponse = (
    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()
)

Manipulating aggregation results

Next, you can extrapolate the spend for each department for the next six months. For this example, you can take a simple approach of using the final month's value as the estimate for the next six months.

current_buckets = ThreeDimensionalAggregation.from_osdk(result, "departmentName", "date")
modified_buckets: list[NestedBucket[str, Range[Date], Double]] = []
date_format = "%Y-%m-%d"
for bucket in current_buckets.buckets:
    # Find the bucket corresponding to the most recent month
    last_bucket: SingleBucket[Date, Double] = bucket[-1].value

    next_six_months: list[SingleBucket[Range[Date], Double]] = []
    # The `date` field has been converted to a string formatted YYYY-MM-DD.
    # Convert to type `Date` from the string. Convert back to a string when
    # creating a SingleBucket object for each month
    current_month: Date = datetime.strptime(last_bucket.key, date_format).date()

    # Loop six times
    for _ in range(6):
        # Construct the next month from the current month
        next_month = current_month + relativedelta(months=1)
        # Add a new bucket which uses the next month as the date range
        # and the most recent months amount as the value
        next_six_months.append(SingleBucket(Range(min=current_month, max=next_month), last_bucket.value))
        current_month = next_month

    # Append the modified results as a NestedBucket
    modified_buckets.append(NestedBucket(bucket.key, next_six_months))

Returning the aggregation

Now that you have created an estimate for the next six months, you can return these estimated values as a ThreeDimensionalAggregation:

return ThreeDimensionalAggregation(modified_buckets)

The full example code for this function is as follows:

from dateutil.relativedelta import relativedelta
from datetime import datetime

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


@function
def estimated_department_expenses() -> ThreeDimensionalAggregation[str, 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_buckets = ThreeDimensionalAggregation.from_osdk(result, "departmentName", "date")
    modified_buckets: list[NestedBucket[str, Range[Date], Double]] = []
    date_format = "%Y-%m-%d"
    for bucket in current_buckets.buckets:
        # Find the bucket corresponding to the most recent month
        last_bucket: SingleBucket[Date, Double] = bucket.buckets[-1]

        next_six_months: list[SingleBucket[Range[Date], Double]] = []
        # The `date` field has been converted to a string formatted YYYY-MM-DD.
        # Convert to type `Date` from the string.
        current_month: Date = datetime.strptime(last_bucket.key, date_format).date()

        # Loop six times
        for _ in range(6):
            # Construct the next month from the current month
            next_month = current_month + relativedelta(months=1)
            # Add a new bucket which uses the next month as the date range
            # and the most recent months amount as the value
            next_six_months.append(SingleBucket(Range(min=current_month, max=next_month), last_bucket.value))
            current_month = next_month

        # Append the modified results as a NestedBucket
        modified_buckets.append(NestedBucket(bucket.key, next_six_months))

    return ThreeDimensionalAggregation(modified_buckets)

The resulting aggregation can be used in a Workshop chart to show the monthly spend estimate for the next six months.

---

中文翻译

创建自定义聚合

可利用函数基于本体（ontology）中的数据计算自定义聚合，并在 Workshop 的图表组件中展示。本指南将介绍如何编写自定义聚合逻辑：从本体加载聚合数据，对结果进行处理以生成未来结果的预测，并返回修改后的结果。

在阅读本节内容时，以下参考资料可能对您有所帮助：

• Python 聚合类型参考文档

:::callout{theme="warning"} TwoDimensionalAggregation 和 ThreeDimensionalAggregation 中的 .from_osdk() 仅在 Python OSDK v2 版本中受支持。 :::

加载聚合

在本示例中，假设您有一个包含费用数据的本体，每个 expense 对象具有部门名称、费用日期和费用金额等属性。如果您需要估算未来六个月内各部门的月度支出，可以首先加载月度支出的聚合数据：

client = FoundryClient()
result: AggregateObjectsResponse = (
    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_buckets = ThreeDimensionalAggregation.from_osdk(result, "departmentName", "date")
modified_buckets: list[NestedBucket[str, Range[Date], Double]] = []
date_format = "%Y-%m-%d"
for bucket in current_buckets.buckets:
    # 查找对应最近月份的桶
    last_bucket: SingleBucket[Date, Double] = bucket[-1].value

    next_six_months: list[SingleBucket[Range[Date], Double]] = []
    # `date` 字段已转换为格式为 YYYY-MM-DD 的字符串。
    # 从字符串转换为 `Date` 类型。在创建每个月的 SingleBucket 对象时再转换回字符串
    current_month: Date = datetime.strptime(last_bucket.key, date_format).date()

    # 循环六次
    for _ in range(6):
        # 从当前月份构造下一个月
        next_month = current_month + relativedelta(months=1)
        # 添加新的桶，使用下个月作为日期范围
        # 并使用最近月份的金额作为值
        next_six_months.append(SingleBucket(Range(min=current_month, max=next_month), last_bucket.value))
        current_month = next_month

    # 将修改后的结果作为 NestedBucket 追加
    modified_buckets.append(NestedBucket(bucket.key, next_six_months))

返回聚合结果

现在您已经创建了未来六个月的估算值，可以将这些估算值作为 ThreeDimensionalAggregation 返回：

return ThreeDimensionalAggregation(modified_buckets)

该函数的完整示例代码如下：

from dateutil.relativedelta import relativedelta
from datetime import datetime

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


@function
def estimated_department_expenses() -> ThreeDimensionalAggregation[str, 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_buckets = ThreeDimensionalAggregation.from_osdk(result, "departmentName", "date")
    modified_buckets: list[NestedBucket[str, Range[Date], Double]] = []
    date_format = "%Y-%m-%d"
    for bucket in current_buckets.buckets:
        # 查找对应最近月份的桶
        last_bucket: SingleBucket[Date, Double] = bucket.buckets[-1]

        next_six_months: list[SingleBucket[Range[Date], Double]] = []
        # `date` 字段已转换为格式为 YYYY-MM-DD 的字符串。
        # 从字符串转换为 `Date` 类型。
        current_month: Date = datetime.strptime(last_bucket.key, date_format).date()

        # 循环六次
        for _ in range(6):
            # 从当前月份构造下一个月
            next_month = current_month + relativedelta(months=1)
            # 添加新的桶，使用下个月作为日期范围
            # 并使用最近月份的金额作为值
            next_six_months.append(SingleBucket(Range(min=current_month, max=next_month), last_bucket.value))
            current_month = next_month

        # 将修改后的结果作为 NestedBucket 追加
        modified_buckets.append(NestedBucket(bucket.key, next_six_months))

    return ThreeDimensionalAggregation(modified_buckets)

生成的聚合结果可在 Workshop 图表中使用，以展示未来六个月的月度支出估算。