“

Series: S12 Developer Tutorials · Article 9 | Level: Intermediate | Reading Time: 15 min

Metric Development Guide

#TL;DR

Metrics are the core data units for measuring business performance in the coomia-dip platform. This guide covers how to define, compute, store, and visualize business metrics, including three types: base metrics, derived metrics, and composite metrics. It covers YAML declaration, Python custom computation, real-time/batch computation modes, and metric monitoring and alerting.

#1. Metric System Overview

#1.1 What is a Metric?

A Metric is a quantitative measurement of Ontology object properties. The coomia-dip metric system runs on the Control Layer (Control Layer) and Data Layer (Data Layer), served via gRPC.

Code

Metric Definition (Schema)
    │
    ▼
Metric Computation (Engine)
    ├── Real-time (Streaming)
    ├── Batch
    └── On-Demand
    │
    ▼
Metric Storage (Doris OLAP)
    │
    ▼
Metric Consumption (Dashboard / API / Alert)

#1.2 Metric Types

Type	Description	Examples
Base	Direct data aggregation	Total revenue, customer count
Derived	Computed from base metrics	Average order value = Revenue / Orders
Composite	Multi-dimensional combination	Revenue per employee per department

#1.3 Metric Dimensions

Dimensions are the analytical facets of metrics:

YAML

dimensions:
  - time: [day, week, month, quarter, year]
  - region: [country, province, city]
  - department: [business_unit, team]
  - product: [category, subcategory]
  - customer: [tier, industry, region]

#2. Environment Setup

Python

from ontology_sdk import OntoPlatform

platform = OntoPlatform(
    control_plane_url="localhost:50051",
    data_plane_url="localhost:50052"
)

metric_manager = platform.metrics

#3. Defining Base Metrics

#3.1 YAML Declaration

YAML

# metrics/total_revenue.yaml
name: total_revenue
display_name: Total Revenue
description: Total amount of all completed orders
category: financial
unit: USD
precision: 2

source:
  object_type: Order
  filter:
    status: completed
  aggregation:
    function: SUM
    field: total_amount

dimensions:
  - name: time
    field: completed_at
    granularities: [day, week, month, quarter, year]
  - name: region
    field: customer.region
  - name: product_category
    field: items.category

schedule:
  compute_mode: batch
  cron: "0 1 * * *"
  timezone: UTC

cache:
  ttl: 3600
  warm_on_compute: true

alerts:
  - name: revenue_drop
    condition: "current < previous * 0.8"
    comparison: day_over_day
    channel: slack
    message: "Daily revenue dropped more than 20% compared to yesterday"

#3.2 Python API Definition

Python

from ontology_sdk.metrics import MetricBuilder, Aggregation, Dimension

metric = (
    MetricBuilder("total_revenue")
    .display_name("Total Revenue")
    .description("Total amount of all completed orders")
    .category("financial")
    .unit("USD")
    .precision(2)
    .source(
        object_type="Order",
        filter={"status": "completed"},
        aggregation=Aggregation.SUM("total_amount")
    )
    .dimension(Dimension.time("completed_at", ["day", "week", "month", "quarter", "year"]))
    .dimension(Dimension.category("customer.region", name="region"))
    .dimension(Dimension.category("items.category", name="product_category"))
    .schedule(compute_mode="batch", cron="0 1 * * *")
    .cache(ttl=3600)
    .build()
)

metric_manager.register(metric)

#3.3 More Base Metrics

Python

# Active customer count
customer_count = (
    MetricBuilder("active_customer_count")
    .display_name("Active Customer Count")
    .source(
        object_type="Customer",
        filter={"status": "active"},
        aggregation=Aggregation.COUNT()
    )
    .dimension(Dimension.category("tier"))
    .dimension(Dimension.category("industry"))
    .dimension(Dimension.time("last_active_at", ["month", "quarter"]))
    .build()
)

# Average order amount
avg_order_amount = (
    MetricBuilder("avg_order_amount")
    .display_name("Average Order Amount")
    .source(
        object_type="Order",
        filter={"status": "completed"},
        aggregation=Aggregation.AVG("total_amount")
    )
    .dimension(Dimension.time("completed_at", ["day", "month"]))
    .build()
)

# Order count
order_count = (
    MetricBuilder("order_count")
    .display_name("Order Count")
    .source(
        object_type="Order",
        filter={"status": "completed"},
        aggregation=Aggregation.COUNT()
    )
    .dimension(Dimension.time("completed_at", ["day", "month"]))
    .build()
)

metric_manager.register_batch([customer_count, avg_order_amount, order_count])

#4. Defining Derived Metrics

#4.1 Formula-Based Derivation

YAML

# metrics/customer_unit_price.yaml
name: customer_unit_price
display_name: Customer Unit Price
description: Average spending per active customer
category: financial
type: derived

formula:
  expression: "total_revenue / active_customer_count"
  dependencies:
    - total_revenue
    - active_customer_count

dimensions:
  - name: time
    granularities: [month, quarter]
  - name: region

Python

from ontology_sdk.metrics import DerivedMetricBuilder

customer_unit_price = (
    DerivedMetricBuilder("customer_unit_price")
    .display_name("Customer Unit Price")
    .formula("total_revenue / active_customer_count")
    .dependencies(["total_revenue", "active_customer_count"])
    .unit("USD")
    .precision(2)
    .build()
)

metric_manager.register(customer_unit_price)

#4.2 Complex Formulas

Python

# Gross margin rate
gross_margin = (
    DerivedMetricBuilder("gross_margin_rate")
    .display_name("Gross Margin Rate")
    .formula("(total_revenue - total_cost) / total_revenue * 100")
    .dependencies(["total_revenue", "total_cost"])
    .unit("%")
    .precision(1)
    .build()
)

# Year-over-year growth rate
yoy_growth = (
    DerivedMetricBuilder("revenue_yoy_growth")
    .display_name("Revenue YoY Growth")
    .formula("(current_period - same_period_last_year) / same_period_last_year * 100")
    .time_comparison(
        current="total_revenue",
        offset="1 year",
        granularity="month"
    )
    .unit("%")
    .precision(1)
    .build()
)

# Month-over-month growth rate
mom_growth = (
    DerivedMetricBuilder("revenue_mom_growth")
    .display_name("Revenue MoM Growth")
    .formula("(current_period - previous_period) / previous_period * 100")
    .time_comparison(
        current="total_revenue",
        offset="1 month",
        granularity="month"
    )
    .unit("%")
    .precision(1)
    .build()
)

#5. Custom Computation Logic

#5.1 Python Custom Metrics

Python

from ontology_sdk.metrics import custom_metric, MetricContext
from pydantic import BaseModel


class HealthScoreResult(BaseModel):
    score: float
    grade: str
    factors: dict


@custom_metric(
    name="project_health_score",
    display_name="Project Health Score",
    description="Multi-dimensional project health assessment",
    compute_mode="on_demand",
    cache_ttl=1800,
)
def compute_project_health(ctx: MetricContext) -> list[HealthScoreResult]:
    """Compute health scores for all in-progress projects"""

    projects = ctx.oql.execute("""
        FIND Project
        WHERE status = 'in_progress'
        INCLUDE
            TRAVERSE has_task -> Task
            AGGREGATE
                COUNT(*) AS total_tasks,
                COUNT(CASE WHEN Task.status = 'done' THEN 1 END) AS done_tasks,
                COUNT(CASE WHEN Task.due_date < NOW() AND Task.status != 'done' THEN 1 END) AS overdue_tasks
        SELECT name, budget, start_date, end_date, total_tasks, done_tasks, overdue_tasks
    """)

    results = []
    for project in projects:
        # Progress factor (0-30 points)
        completion = project.done_tasks / max(project.total_tasks, 1)
        time_elapsed = (ctx.now() - project.start_date).days
        expected_duration = (project.end_date - project.start_date).days
        time_ratio = time_elapsed / max(expected_duration, 1)
        progress_score = max(0, 30 * (1 - abs(completion - time_ratio)))

        # Overdue factor (0-30 points)
        overdue_ratio = project.overdue_tasks / max(project.total_tasks, 1)
        overdue_score = 30 * (1 - overdue_ratio)

        # Budget factor (0-20 points)
        budget_used = ctx.get_metric("project_budget_used", filter={"project_rid": project.rid})
        budget_ratio = budget_used / max(project.budget, 1) if budget_used else 0
        budget_score = 20 if budget_ratio < time_ratio * 1.1 else max(0, 20 * (1 - (budget_ratio - time_ratio)))

        # Team factor (0-20 points)
        team_velocity = ctx.get_metric("team_velocity", filter={"project_rid": project.rid})
        team_score = min(20, 20 * (team_velocity / 10)) if team_velocity else 10

        total_score = progress_score + overdue_score + budget_score + team_score
        grade = "A" if total_score >= 85 else "B" if total_score >= 70 else "C" if total_score >= 55 else "D" if total_score >= 40 else "F"

        results.append(HealthScoreResult(
            score=round(total_score, 1),
            grade=grade,
            factors={
                "progress": round(progress_score, 1),
                "overdue": round(overdue_score, 1),
                "budget": round(budget_score, 1),
                "team": round(team_score, 1),
            }
        ))

    return results

#6. Querying and Consuming Metrics

#6.1 Basic Queries

Python

# Single dimension query
revenue = metric_manager.query(
    "total_revenue",
    time_range=("2025-01-01", "2025-12-31"),
    granularity="month"
)

for point in revenue.data_points:
    print(f"{point.time}: ${point.value:,.2f}")

# Multi-dimensional cross query
breakdown = metric_manager.query(
    "total_revenue",
    time_range=("2025-01-01", "2025-12-31"),
    granularity="month",
    dimensions=["region", "product_category"]
)

for group in breakdown.groups:
    print(f"\n{group.dimension_values}:")
    for point in group.data_points:
        print(f"  {point.time}: ${point.value:,.2f}")

#6.2 Comparison Queries

Python

# Year-over-year comparison
comparison = metric_manager.compare(
    "total_revenue",
    current_period=("2025-01-01", "2025-12-31"),
    previous_period=("2024-01-01", "2024-12-31"),
    granularity="month"
)

for point in comparison.data_points:
    change = ((point.current - point.previous) / point.previous * 100
              if point.previous else 0)
    print(f"{point.time}: ${point.current:,.0f} "
          f"(Last year: ${point.previous:,.0f}, Change: {change:+.1f}%)")

#6.3 Ranking Queries

Python

# Regional revenue ranking
ranking = metric_manager.rank(
    "total_revenue",
    dimension="region",
    time_range=("2025-01-01", "2025-03-31"),
    top_n=10,
    order="desc"
)

for i, entry in enumerate(ranking.entries, 1):
    print(f"  #{i} {entry.dimension_value}: ${entry.value:,.0f}")

#7. Real-Time Metrics

#7.1 Configuring Real-Time Computation

YAML

# metrics/realtime_order_rate.yaml
name: realtime_order_rate
display_name: Real-time Order Rate
type: base
compute_mode: streaming

source:
  type: kafka
  topic: order-events
  filter:
    event_type: order_created
  aggregation:
    function: COUNT
    window:
      type: sliding
      size: 5m
      slide: 1m

dimensions:
  - name: region
    field: customer.region

#7.2 Streaming Metric Consumption

Python

# Subscribe to real-time metrics
async def monitor_orders():
    async for update in metric_manager.subscribe("realtime_order_rate"):
        print(f"[{update.timestamp}] Order rate: {update.value}/min")
        if update.value < 10:
            print("  ALERT: Abnormally low order rate!")

import asyncio
asyncio.run(monitor_orders())

#8. Metric Alerting

#8.1 Alert Rules

Python

from ontology_sdk.metrics import AlertRule, AlertCondition

# Absolute value alert
alert1 = AlertRule(
    name="revenue_floor",
    metric="total_revenue",
    condition=AlertCondition.less_than(100000),
    granularity="day",
    channel="slack",
    recipients=["#sales-alerts"],
    message="Daily revenue dropped below $100,000"
)

# Period-over-period alert
alert2 = AlertRule(
    name="revenue_drop",
    metric="total_revenue",
    condition=AlertCondition.period_over_period_drop(threshold=0.2),
    comparison="day_over_day",
    channel="email",
    recipients=["cfo@company.com"],
    message="Daily revenue dropped more than 20% day-over-day"
)

# Anomaly detection alert
alert3 = AlertRule(
    name="order_anomaly",
    metric="order_count",
    condition=AlertCondition.anomaly_detection(
        method="z_score",
        threshold=3.0,
        training_window="30d"
    ),
    channel="slack",
    recipients=["#ops-alerts"],
    message="Statistical anomaly detected in order volume"
)

metric_manager.register_alerts([alert1, alert2, alert3])

#9. Metric Catalog and Metadata

#9.1 Browsing Metrics

Python

# List all metrics
all_metrics = metric_manager.list(category="financial")
for m in all_metrics:
    print(f"{m.name}: {m.display_name} [{m.type}]")
    print(f"  Unit: {m.unit}, Dimensions: {[d.name for d in m.dimensions]}")

# View metric details
detail = metric_manager.describe("total_revenue")
print(f"Name: {detail.display_name}")
print(f"Description: {detail.description}")
print(f"Computation: {detail.computation}")
print(f"Dependencies: {detail.dependencies}")
print(f"Last computed: {detail.last_computed_at}")
print(f"Freshness: {detail.freshness}")

# Metric lineage
lineage = metric_manager.get_lineage("customer_unit_price")
print("Dependencies:")
for dep in lineage.upstream:
    print(f"  <- {dep.name} ({dep.type})")
print("Dependents:")
for dep in lineage.downstream:
    print(f"  -> {dep.name} ({dep.type})")

#10. Complete Example: Building a Sales Metric System

Python

from ontology_sdk import OntoPlatform
from ontology_sdk.metrics import MetricBuilder, DerivedMetricBuilder, Aggregation, Dimension

platform = OntoPlatform(
    control_plane_url="localhost:50051",
    data_plane_url="localhost:50052"
)
mm = platform.metrics

# === Base Metrics ===
base_metrics = [
    MetricBuilder("total_revenue")
        .display_name("Total Revenue").unit("USD")
        .source("Order", {"status": "completed"}, Aggregation.SUM("total_amount"))
        .dimension(Dimension.time("completed_at", ["day", "month", "quarter", "year"]))
        .dimension(Dimension.category("customer.region", name="region"))
        .build(),

    MetricBuilder("total_cost")
        .display_name("Total Cost").unit("USD")
        .source("Order", {"status": "completed"}, Aggregation.SUM("cost"))
        .dimension(Dimension.time("completed_at", ["day", "month", "quarter"]))
        .build(),

    MetricBuilder("order_count")
        .display_name("Order Count").unit("orders")
        .source("Order", {"status": "completed"}, Aggregation.COUNT())
        .dimension(Dimension.time("completed_at", ["day", "month"]))
        .build(),

    MetricBuilder("new_customer_count")
        .display_name("New Customer Count").unit("customers")
        .source("Customer", {}, Aggregation.COUNT())
        .dimension(Dimension.time("created_at", ["day", "month"]))
        .build(),
]

# === Derived Metrics ===
derived_metrics = [
    DerivedMetricBuilder("gross_profit")
        .display_name("Gross Profit").unit("USD")
        .formula("total_revenue - total_cost")
        .dependencies(["total_revenue", "total_cost"])
        .build(),

    DerivedMetricBuilder("gross_margin_rate")
        .display_name("Gross Margin Rate").unit("%")
        .formula("(total_revenue - total_cost) / total_revenue * 100")
        .dependencies(["total_revenue", "total_cost"])
        .build(),

    DerivedMetricBuilder("avg_order_value")
        .display_name("Average Order Value").unit("USD")
        .formula("total_revenue / order_count")
        .dependencies(["total_revenue", "order_count"])
        .build(),
]

# Batch register
mm.register_batch(base_metrics + derived_metrics)

# === Query Examples ===
print("=== 2025 Monthly Sales Overview ===")
revenue_data = mm.query("total_revenue", time_range=("2025-01-01", "2025-12-31"), granularity="month")
margin_data = mm.query("gross_margin_rate", time_range=("2025-01-01", "2025-12-31"), granularity="month")

for rev, margin in zip(revenue_data.data_points, margin_data.data_points):
    print(f"  {rev.time}: Revenue ${rev.value:,.0f} | Margin {margin.value:.1f}%")

# === Regional Ranking ===
print("\n=== Q1 Regional Revenue Ranking ===")
ranking = mm.rank("total_revenue", dimension="region",
                  time_range=("2025-01-01", "2025-03-31"), top_n=5)
for i, entry in enumerate(ranking.entries, 1):
    print(f"  #{i} {entry.dimension_value}: ${entry.value:,.0f}")

#Key Takeaways

Three-tier metric system: Base metrics aggregate from data, derived metrics combine via formulas, composite metrics enable multi-dimensional analysis
Declarative definition: Prefer YAML for metric definitions; use Python for complex computation logic
Multiple computation modes: Batch for historical stats, streaming for monitoring, on-demand for exploration
Flexible dimensions: Freely cross-analyze across time, region, business type, and other dimensions
Alert-driven: Metric anomalies automatically trigger alerts with absolute value, period-over-period, and anomaly detection methods
Lineage tracking: Metric dependency relationships are visualized with clear upstream/downstream impact

#Next Article

Next: S12-10 Dashboard Development Guide — Learn how to build visual dashboards based on metrics.

Tags: Metrics KPI OLAP Data Analytics Alerting coomia-dip