Finance — portfolio metrics & risk limits

The problem. A trading desk runs several portfolios, each holding positions in many instruments. Risk wants a tidy, queryable home for per-portfolio time-series — mark-to-market value, daily P&L, one-day Value-at-Risk — segmented so each desk sees only its own books, plus an event whenever a portfolio breaches a risk limit.

This scenario leans on datasets to partition data by desk, and on events as a durable record of limit breaches.

Set up demo data

New workspace? Run this once (Python) to create the desk's dataset and a portfolio mark-to-market series with a month of daily values. Safe to re-run.

import datahub_sdk, numpy as np, pandas as pd

client = datahub_sdk.DataHubClient.from_env()

desk_id = client.datasets.create([datahub_sdk.Dataset(external_id="desk_equities", name="Equities desk")])[0].id
client.timeseries.create([datahub_sdk.TimeSeries(external_id="portfolio_growth_mtm_usd",
    name="Growth portfolio — MtM (USD)", unit="usd", value_type="numeric", data_set_id=desk_id)])
idx = pd.date_range(end=pd.Timestamp.now(tz="UTC"), periods=30, freq="1d")
client.timeseries.insert_from_lists(timestamps=idx,
    values=1e7 + np.cumsum(np.random.normal(0, 5e4, 30)), ts="portfolio_growth_mtm_usd")

1. Partition by desk with a dataset

A dataset is the unit of grouping and access. Give each desk its own; everything the desk owns — series, positions — carries its data_set_id.

Java

DataSetModel desk = new DataSetModel();
desk.setExternalId("desk_equities");
desk.setName("Equities desk");
var created = client.datasets().create(List.of(desk));
long deskId = created.getItems().iterator().next().getId();

Python

import datahub_sdk

created = client.datasets.create([
    datahub_sdk.Dataset(external_id="desk_equities", name="Equities desk")])
desk_id = created[0].id

Rust

use dataplatform_rust_sdk::datasets::Dataset;

let mut ds = Dataset::new("Equities desk".into());
ds.external_id = "desk_equities".into();
let created = api.datasets.create(&vec![ds]).await?;
let desk_id = created.get_items()[0].id;

2. Record per-portfolio metrics

Create a series per portfolio metric, tagged with the desk's data_set_id, then append a point at each mark. Values are exact decimals — pick a NUMERIC value type so currency amounts store faithfully.

Java

var mtm = Timeseries.of("portfolio_growth_mtm_usd")
    .name("Growth portfolio — mark-to-market (USD)")
    .setValueType("numeric")
    .setDataSetId(deskId);
mtm.setUnit("usd");
client.timeseries().create(mtm);

client.timeseries().ingest(Map.of(
    "portfolio_growth_mtm_usd", List.of(
        Datapoint.of(Instant.now(), "10428155.42"))));   // exact decimal as text

Python

client.timeseries.create([datahub_sdk.TimeSeries(
    external_id="portfolio_growth_mtm_usd",
    name="Growth portfolio — mark-to-market (USD)",
    unit="usd", value_type="numeric", data_set_id=desk_id)])

client.timeseries.insert_from_lists(
    timestamps=[pd.Timestamp.now(tz="UTC")], values=[10428155.42],
    ts="portfolio_growth_mtm_usd")

Rust

use dataplatform_rust_sdk::timeseries::TimeSeries;
use chrono::Utc;

let mut ts = TimeSeries::new("portfolio_growth_mtm_usd", "Growth portfolio — mark-to-market (USD)");
ts.unit = Some("usd".into());
ts.value_type = "numeric".into();
ts.data_set_id = desk_id;          // desk_id is already Option<u64>
api.time_series.create_one(&ts).await?;

api.time_series
    .insert_datapoint(None, Some("portfolio_growth_mtm_usd".into()), Utc::now(),
                      "10428155.42".into())
    .await?;

3. Flag a risk-limit breach

When a portfolio's one-day VaR exceeds its mandate, record a var_limit_breach event. Because the event carries structured metadata, risk can reconstruct exactly which book breached, by how much, and when.

Java

EventModel breach = new EventModel();
breach.setExternalId("var_limit_breach_growth_" + System.currentTimeMillis());
breach.setType("var_limit_breach");
breach.setStatus("open");
breach.setDataSetId(deskId);
breach.setMetadata(Map.of(
    "portfolio", "portfolio_growth", "var_1d_usd", "512340", "limit_usd", "500000"));
breach.setEventTime(ZonedDateTime.now());
client.events().create(List.of(breach));

Python

client.events.create([datahub_sdk.Event(
    external_id=f"var_limit_breach_growth_{int(pd.Timestamp.now().timestamp())}",
    type="var_limit_breach", status="open", data_set_id=desk_id,
    event_time=pd.Timestamp.now(tz="UTC"),
    metadata={"portfolio": "portfolio_growth",
              "var_1d_usd": "512340", "limit_usd": "500000"})])

Rust

use dataplatform_rust_sdk::events::Event;
use chrono::Utc;

let mut breach = Event::new(format!("var_limit_breach_growth_{}", Utc::now().timestamp()));
breach.r#type = Some("var_limit_breach".into());
breach.status = Some("open".into());
breach.data_set_id = desk_id;
breach.add_metadata("portfolio".into(), "portfolio_growth".into());
breach.add_metadata("var_1d_usd".into(), "512340".into());
breach.add_metadata("limit_usd".into(), "500000".into());
breach.set_event_time(Utc::now());
api.events.create(&vec![breach]).await?;

Exact money

Use the NUMERIC value type for monetary series — it stores exact decimals, unlike the floating-point types. See the Time-series reference for the full list of value types.

See the result

Chart the portfolio's mark-to-market over the month:

import matplotlib.pyplot as plt

rf = datahub_sdk.RetrieveFilter(ts="portfolio_growth_mtm_usd",
    start=pd.Timestamp.now(tz="UTC") - pd.Timedelta(days=30), end=pd.Timestamp.now(tz="UTC"))
s = pd.Series({pd.to_datetime(p.timestamp): float(p.value)
               for p in client.timeseries.retrieve_datapoints(rf)[0].get_datapoints()}).sort_index()
s.plot(title="Growth portfolio mark-to-market (USD)"); plt.show()

See also

• Datasets reference — grouping and access by desk.
• Turn readings into events — the limit-breach rule.
• Query & aggregate — daily P&L roll-ups.