This project is accompanied by a blog paper that covers the design rationale, architecture decisions, and benchmarking methodology behind RiskEngine and ScenarioForge.
RiskEngine & ScenarioForge — Blog Paper (PDF)
The paper walks through the end-to-end pipeline design: synthetic market event generation with ScenarioForge, in-process ONNX inference with Isolation Forest, TimescaleDB persistence, and real-time dashboard observability. It also covers the benchmarking approach and the reasoning behind key architectural trade-offs.
Get the full pipeline running in under 5 minutes.
Prerequisites: Docker Desktop, Java 21, Python 3.12+
# 1. Clone the repo
git clone https://github.com/Fnc-Jit/Risk-An.git
cd Risk-An/risk-engine
# 2. Install Python dependency
pip3 install kafka-python
# 3. Start Kafka, TimescaleDB, and the Java app
docker compose up timescaledb kafka riskengine
# 4. (New terminal) Run ScenarioForge — interactive wizard
python3 run_scenarioforge.py --interactive
# → picks a template, sets EPS/duration, opens dashboard automatically
# 5. Dashboard opens at http://localhost:8050
# API available at http://localhost:8080That's it. The dashboard auto-opens when ScenarioForge starts.
| Requirement | Version | Purpose |
|---|---|---|
| Docker Desktop | 4.x+ | Runs Kafka, TimescaleDB, Java app |
| Java (Temurin) | 21 | Local builds and tests |
| Python | 3.12+ | ScenarioForge event generator |
| kafka-python | 2.x | pip3 install kafka-python |
Java 21 is required for the Gradle toolchain. Install via
brew install --cask temurin@21.
RiskEngine-Java is a real-time financial risk scoring pipeline built in Java 21. It is designed to ingest high-volume synthetic market events, transform them into feature vectors, run low-latency ONNX inference in-process, and persist scored risk signals to TimescaleDB.
ScenarioForge is its companion subsystem—an interactive test case maker and configurable synthetic market event generator. It allows users to define custom market scenarios (e.g., normal day, volatile open, flash crash) through structured configuration. It simulates price, volume, spread, regime shifts, and anomaly injection, publishing reproducible event streams directly to Kafka.
A high-level view showing the end-to-end data flow from ScenarioForge to the RiskEngine-Java pipeline and ultimately the Dashboard.
The system is designed with a clear separation of concerns across ingestion, scoring, storage, and API layers. Data flows continuously from event generation to persistence and visualization in a single streamlined path:
ScenarioForge (your Mac) → Kafka (Docker) → Java app (Docker) → TimescaleDB (Docker) → REST API → Dashboard
- Generation: ScenarioForge generates synthetic market events based on declarative JSON configurations.
- Ingestion: Events are published to the
market-eventstopic in Kafka. - Processing & Inference: The Java Spring Boot app consumes these events, extracts features, and runs them through an ONNX anomaly detection model.
- Persistence: The scored risk signals (label, risk score, inference latency) are persisted into a TimescaleDB hypertable.
- Serving: A Spring Boot REST API exposes the latest signals, metrics, and querying capabilities.
- Dashboard: A frontend dashboard visualizes the live risk signals and performance metrics.
Each incoming market event is transformed into a deterministic, numeric feature vector before inference. The core features extracted include:
- Price z-score relative to a rolling window
- Volume ratio relative to historical averages
- Rolling volatility estimate
- Synthetic bid-ask spread
- Time-of-day encoding (e.g., OPEN, MIDDAY, CLOSE)
- Direction streak or momentum flag
The system utilizes an Isolation Forest anomaly detection model.
- Training Data: The model is trained offline in Python using the synthetic financial anomaly features generated by ScenarioForge. ScenarioForge allows for controlled injection of anomalies (price spikes, volume surges, spread blowouts) to create a robust dataset.
- Formulation: Isolation Forest isolates anomalies by randomly selecting a feature and then randomly selecting a split value between the maximum and minimum values of that selected feature. Since anomalies are "few and different," they are isolated closer to the root of the tree, producing a raw anomaly score.
- Deployment: The trained model is exported to ONNX format. RiskEngine loads this ONNX model at startup and runs in-process inference per event, targeting a p99 latency under 50ms at 1K events/sec. The raw score is then normalized into a 0–100 Risk Score.
The normalized 0–100 risk score is mapped to a human-readable label using configurable thresholds:
| Label | Score Range | Meaning |
|---|---|---|
NORMAL |
0 – 49 | Event falls within expected market behavior |
ELEVATED |
50 – 84 | Unusual activity — increased volatility or volume |
CRITICAL |
85 – 100 | Anomaly detected — price spike, crash, or stress event |
Thresholds are configurable in application.yml:
risk:
thresholds:
elevated: 50 # scores >= 50 become ELEVATED
critical: 85 # scores >= 85 become CRITICALTwo additional statuses exist for non-scored events:
| Status | Meaning |
|---|---|
WARMING_UP |
Ticker has fewer than feature.window.min-samples events — rolling window not yet stable |
FAILED |
Inference threw an exception — event persisted but not scored |
An example of the structured JSON logs emitted by the pipeline as it processes events.
Input: MarketEvent (JSON)
{
"eventId": "evt-12345",
"timestamp": "2026-05-30T10:15:22.124Z",
"ticker": "AAPL",
"price": 211.42,
"volume": 148200,
"volatility": 0.018,
"bidAskSpread": 0.07,
"timeBucket": "OPEN",
"anomalyTag": false
}Output: RiskSignal (TimescaleDB / API)
{
"eventId": "evt-12345",
"timestamp": "2026-05-30T10:15:22.124Z",
"ticker": "AAPL",
"rawScore": -0.41,
"riskScore": 82.6,
"label": "ELEVATED",
"latencyMs": 12.4,
"status": "SCORED"
}
A graph showing the pipeline's event throughput over time, demonstrating resilience and backpressure handling under load.
The project is built with production-grade observability in mind, measuring how the service behaves under increasing load:
- Latency Tracking: Measures
p50,p95, andp99inference latencies for every processed event using rolling percentile trackers. - Throughput Objectives: Targets stable operation at up to 5K–10K events/sec.
- Resilience: If database writes fail, or Kafka experiences lag during heavy loads (e.g., flash crashes), the system applies backpressure and retries without crashing the main inference loop.
Kafka acts as the central message broker connecting the generator to the scoring engine.
- It decouples the Python-based ScenarioForge from the Java-based RiskEngine.
- It provides backpressure management and durability, ensuring that sudden market data bursts (like a simulated flash crash) are queued and processed reliably without crashing the inference service.
- The pipeline utilizes a dedicated
market-eventstopic.
The main view of the real-time observability dashboard, tracking live risk signals as they stream in.
The Dashboard serves as the real-time observability layer of the pipeline.
- It queries the Spring Boot REST API (
/signals/latest,/metrics). - It visualizes critical pipeline health metrics, including throughput (events/sec) and latency percentiles (p50, p95, p99).
- It graphs the incoming risk signals in real-time, allowing users to visually track regime shifts, anomalies, and elevated risk scores as configured in ScenarioForge.
A summary view showing aggregated metrics, recent anomalies, and overall risk distribution.
Detailed tabs allowing users to filter risk signals by ticker, status, or anomaly level.
Drill-down view of a specific risk signal, displaying raw model scores, latency, and event metadata.
The system health tab monitoring database connectivity, Kafka status, and ONNX model state.
Docker runs the infrastructure that the Java app depends on. Without it, the app has nothing to connect to. Docker's job is to give you Kafka and TimescaleDB without needing to install them directly on your Mac.
Here's exactly what each container does in this project:
-
riskengine-kafka— the message broker- Receives events published by ScenarioForge.
- Holds them in the
market-eventstopic. - The Java app reads from this topic continuously.
- Without it: no events reach the Java pipeline at all.
-
riskengine-timescaledb— the time-series database- Stores every scored
RiskSignalrow (eventId, ticker, riskScore, label, latencyMs, status). - The Java app writes to it after every inference call.
- The REST API (
/signals/latest,/signals?ticker=...) queries it. - Without it: the Java app crashes on startup (DB connection required).
- Stores every scored
-
riskengine-app— the Java Spring Boot application- Consumes events from Kafka.
- Runs ONNX inference on each event.
- Writes scored signals to TimescaleDB.
- Serves the REST API on
localhost:8080. - Without it: no scoring, no API, no dashboard data.
-
riskengine-generator(optional, inside Docker)- The old basic event generator — replaced by ScenarioForge when you run it locally.
- Still runs inside Docker as a fallback if you don't run ScenarioForge manually.
One command starts the backend infrastructure (Kafka, TimescaleDB, and the Java RiskEngine):
docker compose -f risk-engine/docker-compose.yml up timescaledb kafka riskengineNetworking Details:
- ScenarioForge runs outside Docker on your Mac and connects to Kafka via port 9094 (the external listener).
- The Java app (
riskengine-app) connects to Kafka via port 9092 (the internal Docker network).
The ScenarioForge tool running, generating synthetic market events according to the defined configurations.
Once the infrastructure is up and running, you can launch ScenarioForge locally to start generating events and publishing them to Kafka.
The CLI interactive wizard allowing users to dynamically configure seeds, templates, and EPS on the fly.
python3 run_scenarioforge.py --interactiveThe wizard asks for: starting template, scenario name, seed, target EPS, stop mode (duration or max events), optional custom tickers, anomaly rate/severity/labels, and output mode (kafka/file/console). It prints a summary, asks for confirmation, then runs.
A printed summary of the generated scenario before execution begins.
python3 run_scenarioforge.py --template mixed_demo --preview
# Available templates:
# normal_day | volatile_open | flash_crash | low_liquidity | mixed_demo | benchmark_high_throughputpython3 run_scenarioforge.py --config scenarioforge/config/scenarios/mixed-demo.json --preview
python3 run_scenarioforge.py --config /path/to/your-config.json# Override EPS, duration, seed
python3 run_scenarioforge.py --template mixed_demo --eps 500 --duration 30 --preview
# Write to a file instead of Kafka
python3 run_scenarioforge.py --template mixed_demo --mode file --output events.jsonl
# Publish to Kafka (use port 9094 from the host, 9092 inside Docker)
python3 run_scenarioforge.py --template benchmark_high_throughput --eps 1000 --brokers localhost:9094| Flag | Description |
|---|---|
--interactive, -i |
Interactive wizard |
--template NAME |
Use a built-in template |
--config PATH |
Load a JSON scenario config |
--preview |
Console preview mode (shorthand for --mode console) |
--eps N |
Override target events per second |
--duration N |
Override run duration in seconds |
--seed N |
Override random seed |
--mode kafka|file|console |
Override output mode |
--brokers HOST:PORT |
Override Kafka brokers |
--topic NAME |
Override Kafka topic |
--output PATH |
Output file when --mode file |
- Backend: Java 21, Spring Boot
- Machine Learning: ONNX Runtime (Isolation Forest anomaly detection)
- Message Broker: Apache Kafka
- Database: TimescaleDB (PostgreSQL extension for time-series data)
- Data Generation: Python 3 (ScenarioForge)
- Infrastructure: Docker, Docker Compose
risk-engine/
├── src/main/java/com/jitraj/riskengine/
│ ├── config/ # Externalized config classes (thresholds, ONNX path, etc.)
│ ├── consumer/ # Kafka listener — deserializes and validates MarketEvent
│ ├── controller/ # REST API controllers (no business logic)
│ ├── dto/ # MarketEvent, RiskSignalDto, MetricsResponse, HealthResponse
│ ├── metrics/ # HdrHistogram-backed p50/p95/p99 latency + counters
│ ├── model/ # RiskSignal entity, RiskLabel enum, ProcessingStatus enum
│ ├── repository/ # TimescaleDB writes (batched, idempotent ON CONFLICT)
│ ├── scoring/ # ONNX session lifecycle + inference
│ ├── service/ # FeatureExtractor, RiskSignalMapper, EventProcessingService
│ ├── util/ # RollingWindow for per-ticker state
│ └── benchmark/ # BenchmarkHarness Gradle task
├── src/main/resources/
│ ├── application.yml # All tunable parameters (thresholds, batch sizes, etc.)
│ └── db/init.sql # TimescaleDB hypertable + index creation
├── src/test/ # JUnit 5 + Mockito + Testcontainers test suite
├── scenarioforge/ # Configurable event generator (Python package)
│ ├── main.py # CLI entry point (--interactive, --template, --config)
│ ├── interactive.py # Step-by-step wizard
│ ├── templates.py # 6 built-in scenario templates
│ ├── dashboard_server.py # Proxy server — serves dashboard, avoids CORS
│ ├── config/ # Loader, validator, schema, sample JSON scenarios
│ ├── engine/ # Price/volume/spread simulators, regime scheduler, anomaly injector
│ └── output/ # Kafka / file / console sinks
├── model/
│ ├── risk_model.onnx # Trained Isolation Forest (720KB)
│ └── golden_vectors.json # Parity test vectors (Python ↔ Java)
├── asset/ # README screenshots
├── docs/hardware.md # Benchmark hardware template
├── run_scenarioforge.py # Launcher — works from any directory
├── riskengine-dashboard.html # Frontend dashboard
├── docker-compose.yml # Full local stack
├── Dockerfile # Multi-stage Java build
└── build.gradle # Gradle build (Java 21 toolchain)
The Java Spring Boot application exposes the following REST API endpoints:
| Endpoint | Method | Description | Query Parameters |
|---|---|---|---|
/health |
GET |
Returns service availability, DB connectivity (state health), and ONNX model load status. | None |
/signals/latest |
GET |
Returns the latest N scored signals (default 100). | limit (optional) |
/signals |
GET |
Returns risk signals filtered by ticker and time range. | ticker (required), from, to, limit (optional) |
/signals/{eventId} |
GET |
Returns the specific scored event by its unique ID. | None |
/metrics |
GET |
Returns internal metrics: events consumed/scored, failures, p50/p95/p99 latency, throughput, and DB write count. | None |
The RiskEngine testing suite is designed to ensure the reliability, performance, and correctness of the event-driven risk scoring pipeline. It employs a multi-tiered testing strategy ranging from fast unit tests to full-stack integration tests.
- JUnit 5: The core testing framework for Java.
- Mockito: Used for mocking external dependencies (e.g., ONNX Runtime, Kafka producers) in unit tests.
- Spring Boot Test: For API and application context testing (
@SpringBootTest,@WebMvcTest). - Testcontainers: Spools up lightweight, throwaway instances of Kafka and TimescaleDB in Docker for true end-to-end testing without external dependencies.
- ScenarioForge: Serves as the robust test-data generator to produce complex synthetic market conditions (e.g., flash crashes) for integration testing.
- Unit Tests: Validate JSON deserialization (handling malformed data gracefully), feature extraction (deterministic numeric vectors), and risk score normalization (translating scores into 0-100 scale and correct labels).
- Service & Mock Tests: Use Mockito to mock the ONNX Runtime session, ensuring that the scoring service handles successful inferences and exceptions without requiring the physical
.onnxmodel file in memory during the test. - Integration Tests: Connect to test database instances (via Testcontainers) to ensure that the TimescaleDB hypertable schemas are correct, and that
RiskSignalrecords are inserted, batched, and queried successfully. End-to-End tests simulate events arriving at Kafka and passing through the entire pipeline. - API Tests: Use Spring's
MockMvcto hit REST endpoints, verifying HTTP status codes, JSON responses, and error handling.
# Run all unit + mock tests (fast, no Docker needed)
./gradlew test
# Force re-run even if nothing changed
./gradlew cleanTest test
# Run only a specific test class
./gradlew test --tests "*FeatureExtractorTest"
# Run repository integration tests (requires Docker for Testcontainers)
./gradlew test --tests "*SignalRepositoryIntegrationTest"Note: Set
JAVA_HOMEto Java 21 if you have multiple JDKs:JAVA_HOME=$(/usr/libexec/java_home -v 21) ./gradlew test
Run benchmarks with:
./gradlew benchmark -PtargetRate=1000 -Pseed=42
./gradlew benchmark -PtargetRate=5000 -Pseed=42Results below are from a local Docker Compose stack. See docs/hardware.md for hardware specs.
| Throughput | p50 (ms) | p95 (ms) | p99 (ms) | Error Rate | Notes |
|---|---|---|---|---|---|
| 1K eps | — | — | — | — | Run benchmark to populate |
| 5K eps | — | — | — | — | Run benchmark to populate |
Fill in after running ./gradlew benchmark and record your hardware in docs/hardware.md.
The testing kit is designed to be fully automated via GitHub Actions. On every push or pull request to the main branch, the CI pipeline automatically builds the Java app, runs all Unit/Service/Mock tests, spools up Testcontainers for Integration tests, and blocks merges if any test fails.