Ingestion Service - sgajbi/portfolio-analytics-system GitHub Wiki

Overview

The Ingestion Service is the sole entry point for all incoming data into the Portfolio Analytics System. It provides a suite of REST API endpoints, validates incoming data payloads, and publishes the raw data as events to Kafka for asynchronous processing by downstream services.

Key Responsibilities

  • Provide robust FastAPI endpoints to accept various types of financial data, including transactions, instruments, market prices, and FX rates.
  • Perform initial validation on the structure and format of incoming data payloads.
  • Publish the validated raw data as events to dedicated Kafka topics (e.g., raw_transactions, instruments, market_prices).
  • Serve as a secure gateway that decouples data producers from the internal processing pipeline.

Technology Stack

  • FastAPI (for the REST API framework)
  • Pydantic (for data validation)
  • confluent-kafka-python (for publishing events to Kafka)

API Endpoints

The service exposes the following primary endpoints under the /ingest/ path:

  • POST /transactions: Accepts a list of financial transactions.
  • POST /instruments: Accepts a list of financial instruments.
  • POST /market-prices: Accepts a list of daily market prices.
  • POST /fx-rates: Accepts a list of foreign exchange rates.
  • GET /health: A health check endpoint.

Testing

  • Unit tests validate payload models (DTOs).
  • Integration tests confirm that a successful API call results in a correctly formatted message being published to the corresponding Kafka topic.