Market ingest

Persistence of official market publications (BEAC / BVMAC bulletins, series observations, primary-market auction results, fitted curves) into the tenant SoR. Technical documentation §10.11 describes the target behaviours; this page maps them to the shipped ingest path.

API

POST /api/market/ingest — SHA-256 bulletin dedupe per tenant. PLATFORM_OPERATOR may set organizationId to ingest for another org.

Aspect	Behaviour
Controller	MarketController.Ingest (api/market/ingest)
Permission	financial:write or tenant:manage (RequireAnyPermission)
Body	{ bulletins?, observations?, auctions?, curves?, organizationId? } — at least one collection non-empty
Limits	1000 rows per collection (400 beyond); 400 Nothing to ingest… when all empty
Required fields	bulletins[i].title, observations[i].seriesKey, auctions[i].tenor → 400 Invalid ingest payload with details[]
Org override	organizationId ≠ caller org → 403 PERMISSION_DENIED unless IsPlatformOperator; unknown org → 404
Success	201 { organizationId, bulletins, skippedBulletins, observations, auctions, curves }
Audit	MARKET_INGESTED (entityType=MARKET, entityId=orgId)
BFF	POST /api/locus/market/ingest, validated by MarketIngestSchema (sor-write-schemas.ts)
Read	GET /api/market/overview (financial:read) → bulletins, observations, auctions, curves, blotter, alerts for the tenant

Payload schema

Field names follow MarketIngestSchema; values below are illustrative (the bulletins + observations example is the one in the producer docstring).

{
  "bulletins": [
    { "title": "BEAC bulletin 2026-10", "source": "BEAC", "sha256Hash": "<64 hex>",
      "publishedAt": "2026-10-01T00:00:00Z", "payload": { "url": "https://www.beac.int/..." } }
  ],
  "observations": [
    { "seriesKey": "TIAO", "country": "CM", "value": 3.5, "observedAt": "2026-10-01T00:00:00Z", "meta": {} }
  ],
  "auctions": [
    { "country": "CM", "instrumentType": "BTA", "tenor": "52W", "marginalYield": 4.1,
      "auctionDate": "2026-10-01T00:00:00Z", "payload": {} }
  ],
  "curves": [
    { "country": "CM", "model": "NelsonSiegel", "rmse": 0.12, "publishedAt": "2026-10-01T00:00:00Z",
      "parameters": { "beta0": 0, "beta1": 0, "beta2": 0, "tau1": 0 } }
  ],
  "organizationId": "org_… (PLATFORM_OPERATOR only)"
}

Defaults and clipping (SoR)

Collection	Table	Defaults / constraints
bulletins	MarketBulletins	Source default BEAC (≤ 64), Title ≤ 255, Sha256Hash ≤ 64, PublishedAt default now
observations	MarketObservations	Country default CM (≤ 5, upper-cased), SeriesKey ≤ 64, append-only
auctions	AuctionLines	InstrumentType default BTA (≤ 32), Tenor ≤ 32, MarginalYield, append-only
curves	CurvePublications	Model default NelsonSiegel, ParametersJson, Rmse, append-only

Idempotency: only bulletins are deduplicated — a bulletin whose non-empty Sha256Hash already exists for the target org is counted in skippedBulletins. Observations, auctions and curves are append-only, so a re-run of a cron job duplicates them unless the producer filters upstream.

Producers

Path	Role
Ops UI /market	Paste/upload JSON
locus-workers/scripts/market_ingest_producer.py	CLI / cron
.github/workflows/market-ingest.yml	workflow_dispatch only

Deferred

Automatic download/parse of live BEAC/BVMAC bulletin URLs is still open; the ingest API and ops UI are ready.

Ops UI (locus-ops /market)

MarketIngestAdminView → ops BFF POST /api/market/ingest → SoR with the PLATFORM_OPERATOR session; the operator sets the target organizationId. See Market ingest (ops).

CLI producer

market_ingest_producer.py POSTs to {LOCUS_API_BASE_URL}/api/market/ingest with Authorization: Bearer <LOCUS_API_TOKEN> and X-Organization-Id: <LOCUS_ORGANIZATION_ID>; it mirrors the MarketIngestSchema checks locally.

Flag / env	Meaning
bulletin_file (arg)	JSON path; - or omitted = stdin
--dry-run	Validate only → prints {"dryRun": true, "valid": true, "counts": {…}}
--target-org ORG	Adds organizationId to the body (PLATFORM_OPERATOR sessions only)
--timeout	HTTP timeout (default 60 s)
LOCUS_API_BASE_URL	default http://localhost:5000
LOCUS_API_TOKEN	required session token
LOCUS_ORGANIZATION_ID	required tenant

Exit codes: 0 success, 1 request/HTTP error, 2 usage/validation error.

python3 locus-workers/scripts/market_ingest_producer.py --dry-run path/to/bulletin.json
cat bulletin.json | python3 locus-workers/scripts/market_ingest_producer.py -
# daily 06:00 UTC
0 6 * * * LOCUS_API_BASE_URL=https://api.example LOCUS_API_TOKEN=... \
  LOCUS_ORGANIZATION_ID=org_... python3 /srv/locus/locus-workers/scripts/market_ingest_producer.py \
  /srv/locus/data/market/latest.json

GitHub workflow

market-ingest.yml is manual only (workflow_dispatch): inputs bulletin_file (default locus-workers/data/market/latest.json) and dry_run (default true); secrets LOCUS_API_BASE_URL, LOCUS_API_TOKEN, LOCUS_ORGANIZATION_ID. For scheduling use cron on a trusted host.

Persistence flow

sequenceDiagram
    autonumber
    participant P as Producer (ops UI / CLI / workflow)
    participant BFF as BFF (ops or web)
    participant SoR as MarketController
    participant DB as Postgres
    P->>BFF: JSON {bulletins, observations, auctions, curves}
    BFF->>BFF: MarketIngestSchema (Zod, ≤ 1000 rows/collection)
    BFF->>SoR: POST /api/market/ingest (Bearer session)
    SoR->>SoR: RequireAnyPermission(financial:write | tenant:manage)
    SoR->>SoR: resolve target org (PLATFORM_OPERATOR override)
    SoR->>DB: load existing bulletin hashes for org
    loop bulletins
        alt hash already present
            SoR->>SoR: skippedBulletins++
        else new
            SoR->>DB: MarketBulletins insert
        end
    end
    SoR->>DB: MarketObservations / AuctionLines / CurvePublications insert
    SoR->>DB: AuditEvent MARKET_INGESTED
    SoR-->>BFF: 201 {counts, skippedBulletins}

Intelligence layer on top (technical documentation §10.11)

Capability	Rule
Bulletin ingestion & quarantine	Official BEAC (CRCT) / BVMAC auction bulletins → SHA-256 fingerprint, idempotent upload; corrupted/malformed bulletins quarantined with diagnostics
Market Radar workflow	z-scores on sovereign/corporate series: (\lvert z\rvert \ge 2.0 \to) WATCH; (\lvert z\rvert \ge 3.0 \to) INVESTIGATION_SIGNAL; (N < 3) or (\sigma = 0 \to) null; analyst annotations link to source bulletins
Morning Market Brief	08:00 GMT+1 brief citing verified database observation ids; entity master synced across blotter and underwriting
Auction ledger & consistency	allotted ≤ tendered; marginal yield ≥ weighted-average yield
Nelson-Siegel publishing	records ((\beta_0,\beta_1,\beta_2,\tau_1)) + RMSE; INSUFFICIENT_OBSERVATIONS when liquid points < 4; < 500 ms
Pricing cockpit & stress	linear/bullet schedules, +100 bp / +250 bp parallel and twist shocks; private positions isolated per tenant book

Verification suite: tests/market_and_fixed_income_persistence.test.ts, market_portfolio_search_rbac.test.ts; ops-side locus-ops/tests/market_ingest_and_change_detector.test.ts. Code: src/domains/market/, src/domains/fixed_income/.