ores.marketdata.market_series

1. Flags
2. Columns
3. SQL
- 3.1. Flags
- 3.2. Checks
4. Insert trigger
- 4.1. Validations
5. C++

A catalog entry for a market data series — it records what is being observed: a yield curve, vol surface, spot rate, fixing index, or similar. Standard temporal reference data; changes infrequently so a regular table with GIST exclusion is appropriate.

The series is identified by oresmd_uri, the series URI of the datum its ORE key names, or the fixing URI of its index name. The URI holds every field of the series' identity, so the table keeps no decomposed copy of the key; series_subclass carries the coarse taxonomy for filtering, and the asset classes the series belongs to live in market_series_asset_classes – a set rather than a column, because a pairwise correlation relates two classes at once.

derivation_kind=/=derivation_config_id=/=derivation_config_version mark whether this series is directly observed (the sentinel OBSERVED) or derived by a named mechanism (e.g. IR_CURVE_BOOTSTRAP, CRM_DERIVATION) – so any published series answers "was this observed or computed, and by what" without guessing from the source tag on its observations. Per-observation lineage (which source series/as-of a specific derived point came from) is a separate concern, tracked by observation_lineage, not this catalog-level marker.

provider_code names the provider whose ids the series carries. A series is one item of one provider held by one party, so two providers that quote the same item keep two series, and a party that switches the source of a provider keeps one series and one history. The observations beneath it record the source that produced each one, for information; the source is never part of the key.

1. Flags

2. Columns

2.1. id

Surrogate UUID uniquely identifying this market series.

2.2. party_id

Party that owns this market series.

Set server-side from the authenticated session. Enforced by RLS.

ctx.generate_uuid()

2.3. oresmd_uri

The oresmd identifier this series is, written as a URI: the canonical form of the key's series part – the key without the point the observation at hand carries – or the index name the row arrived under, and the identity the series is read by. The point belongs to the observation rows beneath this one, so a series with two points carries one identity rather than the identity of whichever point arrived first.

The natural key with party_id and provider_code, and unique per party and provider: one party holds one series per identity and provider. It is the only name the row carries: the registry's series_type, metric and qualifier columns have gone, and a reader that wants the key's parts reads the identity through oresmd_uri_codec rather than off a series.

Every writer supplies one. A caller that leaves it empty stores nothing, because the column is not null and the generated service copies the write's fields straight through – so an update carries the identity it read rather than clearing it.

The test-data generator writes a generic index name, the one oresmd URI that names a series with nothing but a name. Its rows cannot collide on the identity: the party is freshly generated per row, and the party is part of the key.

std::string("oresmd://generic/") + std::string(faker::word::noun()) + "-" +
    std::to_string(idx) + "?type=fixing"

2.4. series_subclass

Subclass within the asset class, as a code from refdata.series_subclass_code (referenced series_subclass_code.code). Validated by ores_refdata_validate_series_subclass_code_fn, the same cross-table reference mechanism the asset-class junction uses, and carried as the code itself: the taxonomy is runtime-managed.

[idx]{
    static constexpr std::string_view subclasses[] = {
        "spot",    "forward", "volatility", "yield",       "basis",
        "fra",     "xccy",    "spread",     "index_credit", "recovery",
        "swap",    "capfloor", "seasonality", "price",     "correlation"};
    return std::string(subclasses[static_cast<std::size_t>(idx) % std::size(subclasses)]);
}()

2.5. provider_code

References provider.code: the provider whose ids the series carries. The natural key with party_id and oresmd_uri: one party holds one series of each provider for each item.

std::string("ore")

2.6. derivation_kind

References derivation_kind.code – whether this series is directly observed (the sentinel OBSERVED, the default) or derived by a named mechanism (IR_CURVE_BOOTSTRAP, CRM_DERIVATION, …). Paired with derivation_config_id=/=derivation_config_version: a row's own derivation_kind is the discriminator that says whether those two fields are meaningful or hold their nil/zero sentinel.

std::string("OBSERVED")

2.7. derivation_config_id

The recipe/config that produced this series, when derived – ores_utility_nil_uuid_fn() (the sentinel, not null) when derivation_kind is 'OBSERVED'. Deliberately not a hard FK: the table this id resolves against depends on derivation_kind (the IR curve bootstrap config for IR_CURVE_BOOTSTRAP, the CRM topology config for CRM_DERIVATION), so it is a soft, self-describing reference.

boost::uuids::nil_uuid()

2.8. derivation_config_version

The derivation config's version this series was produced under – 0 (the sentinel) when derivation_kind is 'OBSERVED'.

3. SQL

3.1. Flags

3.2. Checks

expression
"oresmd_uri" <> ''
"series_subclass" <> ''
("derivation_kind" = 'OBSERVED' and "derivation_config_id" = ores_utility_nil_uuid_fn() and "derivation_config_version" = 0) or ("derivation_kind" <> 'OBSERVED' and "derivation_config_id" <> ores_utility_nil_uuid_fn() and "derivation_config_version" <> 0)

4. Insert trigger

4.1. Validations

column	validation_function
series_subclass	ores_refdata_validate_series_subclass_code_fn
provider_code	ores_marketdata_validate_provider_fn
derivation_kind	ores_refdata_validate_derivation_kind_fn

5. C++

5.1. Flags

5.2. Repository

5.2.1. read_latest_by_key

Reads the latest market series matching oresmd_uri and provider_code. When party_id is non-empty, scopes the result to that party.

The URI and the provider are the identity, so a reader that starts here finds a series without knowing how the registry decomposed its key. The method has no default for the provider: a caller that does not know it cannot name a series.

std::vector<domain::market_series> read_latest_by_key(context ctx,
    const std::string& oresmd_uri, const std::string& provider_code,
    const std::string& party_id = {});

#include <string>

std::vector<domain::market_series>
market_series_repository::read_latest_by_key(context ctx, const std::string& oresmd_uri,
                                             const std::string& provider_code,
                                             const std::string& party_id) {
    static constexpr auto max = db_timestamp::max();
    const auto tid = ctx.tenant_id().to_string();

    if (!party_id.empty()) {
        const auto query = query_builder::read<std::vector<market_series_entity>> |
                           query_builder::where("tenant_id"_c == tid && "party_id"_c == party_id &&
                                 "valid_to"_c == max &&
                                 "provider_code"_c == provider_code &&
                                 "oresmd_uri"_c == oresmd_uri) |
                           query_builder::order_by("id"_c);
        return query_builder::execute_read_query<market_series_entity, domain::market_series>(
            ctx, query,
            [](const auto& entities) { return market_series_mapper::map(entities); },
            lg(), "Reading latest market series by URI, provider and party");
    }

    const auto query = query_builder::read<std::vector<market_series_entity>> |
                       query_builder::where("tenant_id"_c == tid && "valid_to"_c == max &&
                             "provider_code"_c == provider_code &&
                             "oresmd_uri"_c == oresmd_uri) |
                       query_builder::order_by("id"_c);
    return query_builder::execute_read_query<market_series_entity, domain::market_series>(
        ctx, query,
        [](const auto& entities) { return market_series_mapper::map(entities); },
        lg(), "Reading latest market series by URI and provider");
}

5.2.2. read_latest_by_uri

Reads the latest market series of oresmd_uri from every provider. A reader that wants a value and does not care who supplied it, such as the spot of an FX forward, has no provider to name.

std::vector<domain::market_series> read_latest_by_uri(context ctx,
    const std::string& oresmd_uri);

std::vector<domain::market_series>
market_series_repository::read_latest_by_uri(context ctx, const std::string& oresmd_uri) {
    static constexpr auto max = db_timestamp::max();
    const auto tid = ctx.tenant_id().to_string();
    const auto query = query_builder::read<std::vector<market_series_entity>> |
                       query_builder::where("tenant_id"_c == tid && "valid_to"_c == max &&
                             "oresmd_uri"_c == oresmd_uri) |
                       query_builder::order_by("id"_c);
    return query_builder::execute_read_query<market_series_entity, domain::market_series>(
        ctx, query,
        [](const auto& entities) { return market_series_mapper::map(entities); },
        lg(), "Reading latest market series by URI");
}

5.2.3. project_identity_after_write

Projects the identity of every series the generated write path writes, so a series is never committed without the row a query joins on.

The hook sits where the row is written rather than at each call site, because every writer of a series reaches this method: the ingest loop, the curve republish, the import and the service. The projector reads the URI through the codec, so the projection cannot disagree with the identity the ticks land under, and the write joins the transaction the context already carries.

A repository of another entity collapses these markers, so the hook costs every other entity nothing.

#include "ores.marketdata.core/repository/market_series_identity_projector.hpp"

market_series_identity_projector::project(ctx, std::vector<domain::market_series>{t});

market_series_identity_projector::project(ctx, batch);

5.3. Handler: backfill_series_identity

Handles backfill_series_identity_request: projects the identity of every series the tenant holds that has none, so a series written before the projection existed becomes joinable.

This is the repair for rows the write path never saw, a database restored from a dump or a series written before the projection landed. It reads what is already projected rather than re-projecting everything, so a second call finds nothing to do and says so. The work is the projector's project_missing, which the data quality publish handler also calls.

#include "ores.marketdata.core/repository/market_series_identity_projector.hpp"
#include "ores.marketdata.core/repository/market_series_identity_repository.hpp"
#include "ores.marketdata.core/repository/market_series_repository.hpp"
#include <boost/uuid/string_generator.hpp>
#include <boost/uuid/uuid_io.hpp>
#include <optional>
#include <set>
#include <string>
#include <vector>

void backfill_identity(ores::nats::message msg) {
    BOOST_LOG_SEV(market_series_handler_lg(), debug) << "Handling " << msg.subject;
    auto req_ctx_expected = ores::service::service::make_request_context(ctx_, msg, verifier_);
    if (!req_ctx_expected) {
        error_reply(nats_, msg, req_ctx_expected.error());
        return;
    }
    const auto& req_ctx = *req_ctx_expected;
    if (!has_permission(req_ctx, "marketdata::market_series:write")) {
        error_reply(nats_, msg, ores::service::error_code::forbidden);
        return;
    }
    auto req = decode<backfill_series_identity_request>(msg);
    if (!req) {
        BOOST_LOG_SEV(market_series_handler_lg(), warn) << "Failed to decode: " << msg.subject;
        error_reply(nats_, msg, ores::service::error_code::bad_request);
        return;
    }

    backfill_series_identity_response resp;
    try {
        std::optional<boost::uuids::uuid> party;
        if (!req->party_id.empty())
            party = boost::uuids::string_generator{}(req->party_id);

        const auto done =
            repository::market_series_identity_projector::project_missing(req_ctx, party);
        const auto total = done.projected + done.already_projected;

        resp.projected_count = static_cast<int>(done.projected);
        resp.already_projected_count = static_cast<int>(done.already_projected);
        resp.unreadable_count = static_cast<int>(done.unreadable);
        resp.success = true;
        resp.message = "Projected " + std::to_string(done.projected) + " of " +
                       std::to_string(total) + " series, " + std::to_string(done.unreadable) +
                       " with an unreadable URI.";
        BOOST_LOG_SEV(market_series_handler_lg(), info) << msg.subject << ": " << resp.message;
    } catch (const std::exception& e) {
        BOOST_LOG_SEV(market_series_handler_lg(), error)
            << msg.subject << " failed: " << e.what();
        resp.message = e.what();
    }
    reply(nats_, msg, resp);
}

5.4. Domain includes

#include <boost/uuid/uuid.hpp>
#include <boost/uuid/nil_generator.hpp>
#include <string>

5.5. Entity includes

#include <string>

5.6. Handler: export_to_storage

Handles export_market_data_to_storage_request: writes the tenant's market data to object storage as the ORE text an engine run reads, and returns the storage key. Used by the reporting workflow to gather market data for report generation. Not part of the CRUD surface: its messages are declared by ores.marketdata.operations and arrive with the component's generated operations_protocol.hpp.

The engine is handed a run document that names a market data file, and it reads that file and nothing else, so what is stored here is that body. A store of the series rows would be a different kind of thing and the engine could not read it.

#include "ores.marketdata.api/messaging/operations_protocol.hpp"
#include "ores.marketdata.core/messaging/party_required.hpp"
#include "ores.marketdata.core/service/ore_export_service.hpp"
#include "ores.nats/service/nats_client.hpp"
#include "ores.storage.core/net/storage_transfer.hpp"
#include <chrono>

void export_to_storage(ores::nats::message msg, const std::string& http_base_url) {
    [[maybe_unused]] const auto correlation_id =
        ores::service::messaging::log_handler_entry(market_series_handler_lg(), msg);
    auto ctx_expected = ores::service::service::make_request_context(ctx_, msg, verifier_);
    if (!ctx_expected) {
        error_reply(nats_, msg, ctx_expected.error());
        return;
    }
    const auto& ctx = *ctx_expected;
    export_market_data_to_storage_response resp;
    try {
        auto req = decode<export_market_data_to_storage_request>(msg);
        if (!req) {
            resp.message = "Failed to decode request.";
            reply(nats_, msg, resp);
            return;
        }
        if (!names_a_party(ctx)) {
            resp.message = party_required_message();
            reply(nats_, msg, resp);
            return;
        }

        const auto began = std::chrono::steady_clock::now();
        const auto since = [&began] {
            return std::chrono::duration_cast<std::chrono::milliseconds>(
                       std::chrono::steady_clock::now() - began)
                .count();
        };
        BOOST_LOG_SEV(market_series_handler_lg(), info)
            << "export_to_storage: exporting for valuation date '"
            << req->valuation_date.value_or("") << "' to " << req->storage_bucket << "/"
            << req->storage_key;

        service::ore_export_service svc(ctx);
        const auto written = svc.write_all(req->valuation_date, req->max_age_days);
        const auto exported_ms = since();

        ores::storage::net::storage_transfer transfer(http_base_url,
                                                      ores::nats::service::extract_bearer(msg));
        transfer.upload_blob(req->storage_bucket, req->storage_key, written.market_data);
        transfer.upload_blob(req->storage_bucket, req->fixings_storage_key, written.fixings);
        BOOST_LOG_SEV(market_series_handler_lg(), info)
            << "export_to_storage: built the export in " << exported_ms
            << " ms and uploaded it in " << (since() - exported_ms) << " ms";

        resp.success = true;
        resp.series_count = written.series_count;
        resp.carried_forward_count = written.carried_forward_count;
        resp.stale_count = written.stale_count;
        resp.storage_key = req->storage_key;
        resp.fixings_storage_key = req->fixings_storage_key;
        resp.message = "Exported " + std::to_string(written.series_count) +
                       " series as ORE market data to storage.";

        BOOST_LOG_SEV(market_series_handler_lg(), info)
            << "export_to_storage: wrote " << written.market_data.size()
            << " bytes of ORE market data for " << written.series_count << " series ("
            << written.observation_count << " observations, " << written.fixing_count
            << " fixings) to " << req->storage_bucket << "/" << req->storage_key;
    } catch (const std::exception& e) {
        BOOST_LOG_SEV(market_series_handler_lg(), error)
            << msg.subject << " failed: " << e.what();
        resp.message = e.what();
    }
    reply(nats_, msg, resp);
}