SDK to API using smart batching + Export to Excel

.github/dependabot.yml

@@ -8,9 +8,7 @@ updates:
   - package-ecosystem: "uv"
     multi-ecosystem-group: "risk-analyzer-service"
     directory: "/"
-    patterns: ["bigdata-client", "bigdata-research-tools"]
-    allow:
-      - dependency-name: "bigdata-client"
+    patterns: ["*"]
   - package-ecosystem: "docker"
     multi-ecosystem-group: "risk-analyzer-service"
     directory: "/"

.github/workflows/test.yml

@@ -17,7 +17,7 @@ jobs:
       - name: Install uv
         uses: astral-sh/setup-uv@v6
       - name: Install dependencies
-        run: uv sync --locked --dev
+        run: uv sync --dev
       - name: Type check the code
         run: make type-check
       - name: Lint the code
@@ -42,6 +42,6 @@ jobs:
           python-version: ${{ matrix.python-version }}
           enable-cache: true
       - name: Install dependencies
-        run: uv sync --locked --dev
+        run: uv sync --dev
       - name: Test with pytest
         run: make tests

CHANGELOG.md

@@ -5,6 +5,21 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [3.0.0] - 25-08-2026
+
+### Changed
+- Migrated off the deprecated `bigdata-client` / `bigdata-research-tools` SDKs onto the Bigdata.com REST API (`X-API-KEY` / `https://api.bigdata.com`), `bigdata-smart-batching` for search, and direct OpenAI calls for taxonomy generation, chunk labeling, and company summaries.
+- Company universes are now provided as a list of RP entity IDs or an uploaded CSV (`RP_ENTITY_ID`/`RP_COMPANY_ID` + `COMPANY_NAME`, optionally `TICKER`/`SECTOR`/`INDUSTRY`/`COUNTRY`). **Watchlists are no longer supported.**
+- `document_type`, `fiscal_year`, `control_entities`, `frequency`, `batch_size`, and `document_limit` request fields were dropped (no equivalent in the new stack); `document_limit`/`batch_size`/`frequency` are replaced by a single `chunk_percentage` retrieval-volume knob. Added `max_leaf_labels` to cap generated taxonomy size, and renamed `llm_model_config` to `llm_model` (plain string, default `gpt-5.6-luna`).
+- **The report storage schema changed and is not backwards compatible.** Databases created by 2.x keep their old columns and must be removed (or `DB_STRING` pointed at a new database) before starting 3.0; the service now fails at startup with an explicit message instead of returning 500s from `/status/{request_id}`.
+
+### Added
+- `POST /risk-analysis/upload`: submit a risk analysis with a universe CSV instead of a list of RP entity IDs.
+
+### Fixed
+- API keys supplied through a `.env` file are now picked up again. `bigdata-client` used to call `load_dotenv()` on import; without it the service started with unset credentials and failed mid-analysis. Settings read `.env` explicitly and the Bigdata.com and OpenAI clients are constructed with the configured keys instead of relying on `os.environ`.
+- Retrieved chunks are attributed to every company from the universe detected in them, rather than to whichever entity ID happened to sort first.
+
 ## [2.3.11] - 09-06-2026
 
 ### Fixed

README.md

@@ -1,5 +1,5 @@
 # Risk Analyzer with Bigdata.com
-This repository contains a docker image for running a risk analyzing service using Bigdata.com SDK. You can read more on our [docs](https://docs.bigdata.com/use-cases/docker-services/risk-analyzer).
+This repository contains a docker image for running a risk analyzing service using the Bigdata.com REST API (`https://api.bigdata.com`, `X-API-KEY` auth) and OpenAI. You can read more on our [docs](https://docs.bigdata.com/use-cases/docker-services/risk-analyzer).
 
 # How to use?
 The risk analyzing service will allow you to assess and quantify risks related to a specific theme, such as US-China trade relations or supply chain disruptions. It will screen your trading universe and quantify the potential impact of identified risks for each company in the universe.
@@ -10,9 +10,8 @@ The risk analyzing service will allow you to assess and quantify risks related t
     - For more information on how to get an API key, refer to the [Bigdata.com documentation](https://docs.bigdata.com/api-reference/introduction#api-key-beta).
 
 # Quickstart
-To quickly get started, you have two options:
 
-1. **Build and run locally:**
+**Build and run locally:**
 You need to build the docker image first and then run it:
 
 ```bash
@@ -32,16 +31,6 @@ docker run -d \
   bigdata_risk_analyzer
 ```
 
-2. **Run directly from GitHub Container Registry:**
-
-```bash
-docker run -d \
-  --name bigdata_risk_analyzer \
-  -p 8000:8000 \
-  -e BIGDATA_API_KEY=<bigdata-api-key-here> \
-  -e OPENAI_API_KEY=<openai-api-key-here> \
-  ghcr.io/bigdata-com/bigdata-risk-analyzer:latest
-```
 
 This will start the risk analyzer service locally on port 8000. You can then access the service @ `http://localhost:8000/` and the documentation for the API @ `http://localhost:8000/docs`.
 
@@ -55,7 +44,9 @@ We perform a pre-release security scan on our container images to detect vulnera
 
 ## How to analyse a set of companies?
 
-A risk analysis report provides an executive summary of financially relevant information about a set of companies that form your watchlist. You can generate a report either using the UI or programmatically, allowing you to build custom workflows on top of this service.
+A risk analysis report provides an executive summary of financially relevant information about a set of companies in your universe. You can generate a report either using the UI or programmatically, allowing you to build custom workflows on top of this service.
+
+The company universe is provided either as a list of RavenPack (RP) entity IDs, or as an uploaded CSV. **Watchlists (watchlist IDs) are not supported.**
 
 ### Using the UI
 There is a very simple UI available @ `http://localhost:8000/` where you can set your parameters and receive an easy-to-read summary of the analysis.
@@ -64,7 +55,8 @@ There is a very simple UI available @ `http://localhost:8000/` where you can set
 The risk analysis API works asynchronously. You first submit a request to start the analysis, then check the status periodically until completion.
 
 #### Step 1: Submit Risk Analysis Request
-Send a POST request to the `/risk-analysis` endpoint with the required parameters. This will return a `request_id` and queue the analysis for processing:
+
+**Option A — a list of RP entity IDs**, via `POST /risk-analysis`:
 
 ```bash
 curl -X 'POST' \
@@ -74,24 +66,26 @@ curl -X 'POST' \
   -d '{
   "main_theme": "US Import Tariffs against China",
   "focus": "Provide a detailed taxonomy of risks describing how new American import tariffs against China will impact US companies, their operations and strategy. Cover trade-relations risks, foreign market access risks, supply chain risks, US market sales and revenue risks (including price impacts), and intellectual property risks, provide at least 4 sub-scenarios for each risk factor.",
-  "companies": "44118802-9104-4265-b97a-2e6d88d74893",
-  "control_entities": {
-    "place": [
-      "China"
-    ]
-  },
+  "companies": ["D8442A", "228D42", "4A6F00"],
   "start_date": "2024-01-01",
   "end_date": "2024-12-31",
-  "keywords": [
-    "Tariffs"
-  ],
-  "document_type": "TRANSCRIPTS",
-  "fiscal_year": 2024,
-  "frequency": "M"
+  "keywords": ["Tariffs"],
+  "chunk_percentage": 0.05,
+  "max_leaf_labels": 15
 }'
 ```
 
-This will return a response like:
+**Option B — a universe CSV**, via `POST /risk-analysis/upload` (multipart, same fields minus `companies`, sent as a JSON string in the `request` form field). The CSV needs `RP_ENTITY_ID` (alias `RP_COMPANY_ID`) and `COMPANY_NAME` columns; `TICKER`/`SECTOR`/`INDUSTRY`/`COUNTRY` are optional enrichment columns:
+
+```bash
+curl -X 'POST' \
+  'http://localhost:8000/risk-analysis/upload' \
+  -H 'accept: application/json' \
+  -F 'file=@Internal/mag7.csv;type=text/csv' \
+  -F 'request={"main_theme": "US Import Tariffs against China", "focus": "Provide a detailed taxonomy of risks describing how new American import tariffs against China will impact US companies.", "start_date": "2024-01-01", "end_date": "2024-12-31"};type=application/json'
+```
+
+Both endpoints return a response like:
 ```json
 {
   "request_id": "550e8400-e29b-41d4-a716-446655440000",
@@ -139,20 +133,10 @@ curl -X 'POST' \
   -d '{
   "main_theme": "US Import Tariffs against China",
   "focus": "Provide a detailed taxonomy of risks describing how new American import tariffs against China will impact US companies, their operations and strategy. Cover trade-relations risks, foreign market access risks, supply chain risks, US market sales and revenue risks (including price impacts), and intellectual property risks, provide at least 4 sub-scenarios for each risk factor.",
-  "companies": "44118802-9104-4265-b97a-2e6d88d74893",
-  "control_entities": {
-    "place": [
-      "China"
-    ]
-  },
+  "companies": ["D8442A", "228D42", "4A6F00"],
   "start_date": "2024-01-01",
   "end_date": "2024-12-31",
-  "keywords": [
-    "Tariffs"
-  ],
-  "document_type": "TRANSCRIPTS",
-  "fiscal_year": 2024,
-  "frequency": "M"
+  "keywords": ["Tariffs"]
 }'
 
 # Check status using the returned request_id

bigdata_risk_analyzer/api/app.py

@@ -2,17 +2,28 @@
 from typing import Annotated
 from uuid import UUID, uuid4
 
-from bigdata_client import Bigdata
-from bigdata_client.models.search import DocumentType
-from fastapi import BackgroundTasks, Body, Depends, FastAPI, HTTPException, Security
+from fastapi import (
+    BackgroundTasks,
+    Body,
+    Depends,
+    FastAPI,
+    File,
+    Form,
+    HTTPException,
+    Security,
+    UploadFile,
+)
 from fastapi.responses import HTMLResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
+from pydantic import ValidationError
+from sqlalchemy import inspect
 from sqlmodel import Session, SQLModel, create_engine
 
 from bigdata_risk_analyzer import LOG_LEVEL, __version__, logger
 from bigdata_risk_analyzer.api.models import (
-    ExampleWatchlists,
+    EXAMPLE_COMPANY_LISTS,
     RiskAnalysisRequest,
+    RiskAnalysisRequestBase,
     RiskAnalyzerAcceptedResponse,
     RiskAnalyzerStatusResponse,
     WorkflowStatus,
@@ -22,17 +33,44 @@
 from bigdata_risk_analyzer.api.utils import get_example_values_from_schema
 from bigdata_risk_analyzer.models import RiskAnalysisResponse
 from bigdata_risk_analyzer.service import process_request
-from bigdata_risk_analyzer.settings import UNSET, settings
+from bigdata_risk_analyzer.settings import settings
 from bigdata_risk_analyzer.templates import loader
-from bigdata_risk_analyzer.traces import TraceEventName, send_trace
+from bigdata_risk_analyzer.universe import (
+    WATCHLIST_REJECTED_MESSAGE,
+    build_universe_from_ids,
+    load_universe_csv,
+)
 
-BIGDATA: Bigdata | None = None
 engine = create_engine(settings.DB_STRING, echo=LOG_LEVEL == "DEBUG")
 
 
+def check_storage_schema():
+    """Fail fast on a database created by an incompatible earlier release.
+
+    ``create_all`` only creates missing tables, so a database left over from
+    2.x keeps its old columns and every read fails later with an opaque 500.
+    """
+    inspector = inspect(engine)
+    for table in SQLModel.metadata.sorted_tables:
+        if not inspector.has_table(table.name):
+            continue
+        existing_columns = {
+            column["name"] for column in inspector.get_columns(table.name)
+        }
+        missing_columns = {column.name for column in table.columns} - existing_columns
+        if missing_columns:
+            raise RuntimeError(
+                f"Table '{table.name}' in {settings.DB_STRING} is missing columns "
+                f"{sorted(missing_columns)}. The 3.0 schema is not compatible with "
+                "databases created by earlier releases. Remove the old database file "
+                "or point DB_STRING at a new one."
+            )
+
+
 def create_db_and_tables():
     logger.info("Setting up data storage", db_string=settings.DB_STRING)
     SQLModel.metadata.create_all(engine)
+    check_storage_schema()
 
 
 def get_session():
@@ -45,23 +83,8 @@ def get_storage_manager(session: Session = Depends(get_session)) -> StorageManag
 
 
 def lifespan(app: FastAPI):
-    global BIGDATA
     logger.info("Starting Risk Analyzer service")
-
-    # Instantiate Bigdata client
-    BIGDATA = Bigdata(api_key=settings.BIGDATA_API_KEY)
-
-    if settings.BIGDATA_API_KEY != UNSET:
-        send_trace(
-            BIGDATA,
-            event_name=TraceEventName.SERVICE_START,
-            trace={
-                "version": __version__,
-            },
-        )
-
     create_db_and_tables()
-
     yield
 
 
@@ -91,7 +114,7 @@ def health_check():
 async def sample_frontend(_: str = Security(query_scheme)) -> HTMLResponse:
     # Get example values from the schema for all fields
     template_values = get_example_values_from_schema(RiskAnalysisRequest)
-    template_values["example_watchlists"] = list(dict(ExampleWatchlists).values())
+    template_values["example_companies"] = EXAMPLE_COMPANY_LISTS
     template_values["demo_mode"] = settings.DEMO_MODE
     template_values["version"] = f"v{__version__}"
 
@@ -101,31 +124,20 @@ async def sample_frontend(_: str = Security(query_scheme)) -> HTMLResponse:
     )
 
 
-@app.post("/risk-analysis", response_model=RiskAnalysisResponse)
-def analyze_risk(
-    request: Annotated[RiskAnalysisRequest, Body()],
+def _queue_analysis(
+    request: RiskAnalysisRequestBase,
+    universe_df,
     background_tasks: BackgroundTasks,
-    storage_manager: StorageManager = Depends(get_storage_manager),
-    _: str = Security(query_scheme),
+    storage_manager: StorageManager,
 ) -> JSONResponse:
-    """This endpoints starts the generation of therisk analyzer workflow on the background
-    and will return a request_id that can be used to check the status of the request in the
-    `/status/{request_id}` endpoint.
-    Note: for now, it only supports news as document type.
-    """
-    # While we improve the UX of working with several document types with different sets of parameters
-    # we will limit the document type to news
-    DOCUMENT_TYPE = DocumentType.NEWS
-    request.document_type = DOCUMENT_TYPE
-    request_id = uuid4()
-
+    request_id: UUID = uuid4()
     storage_manager.update_status(request_id, WorkflowStatus.QUEUED)
 
     background_tasks.add_task(
         partial(
             process_request,
             request,
-            bigdata=BIGDATA,
+            universe_df=universe_df,
             request_id=request_id,
             storage_manager=storage_manager,
         )
@@ -138,6 +150,68 @@ def analyze_risk(
     )
 
 
+@app.post("/risk-analysis", response_model=RiskAnalysisResponse)
+def analyze_risk(
+    request: Annotated[RiskAnalysisRequest, Body()],
+    background_tasks: BackgroundTasks,
+    storage_manager: StorageManager = Depends(get_storage_manager),
+    _: str = Security(query_scheme),
+) -> JSONResponse:
+    """This endpoint starts the generation of the risk analyzer workflow on the background
+    and will return a request_id that can be used to check the status of the request in the
+    `/status/{request_id}` endpoint.
+
+    `companies` must be a list of RavenPack entity IDs. Watchlists are not supported; upload
+    a CSV via `/risk-analysis/upload` for larger or metadata-rich universes.
+    """
+    companies = request.companies
+    if isinstance(companies, str):
+        raise HTTPException(status_code=400, detail=WATCHLIST_REJECTED_MESSAGE)
+    try:
+        universe_df = build_universe_from_ids(
+            companies, api_key=settings.BIGDATA_API_KEY
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+    return _queue_analysis(request, universe_df, background_tasks, storage_manager)
+
+
+@app.post("/risk-analysis/upload", response_model=RiskAnalysisResponse)
+def analyze_risk_upload(
+    background_tasks: BackgroundTasks,
+    file: Annotated[
+        UploadFile,
+        File(description="Universe CSV with RP_ENTITY_ID + COMPANY_NAME columns."),
+    ],
+    request: Annotated[
+        str,
+        Form(
+            description="JSON-encoded request body (same fields as POST /risk-analysis, minus companies)."
+        ),
+    ],
+    storage_manager: StorageManager = Depends(get_storage_manager),
+    _: str = Security(query_scheme),
+) -> JSONResponse:
+    """Same as `POST /risk-analysis`, but the company universe comes from an uploaded CSV
+    (columns: `RP_ENTITY_ID` [alias `RP_COMPANY_ID`], `COMPANY_NAME`, and optionally
+    `TICKER`/`SECTOR`/`INDUSTRY`/`COUNTRY`) instead of a list of RP entity IDs.
+    """
+    try:
+        parsed_request = RiskAnalysisRequestBase.model_validate_json(request)
+    except ValidationError as e:
+        raise HTTPException(status_code=422, detail=e.errors())
+
+    try:
+        universe_df = load_universe_csv(file.file, api_key=settings.BIGDATA_API_KEY)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+    return _queue_analysis(
+        parsed_request, universe_df, background_tasks, storage_manager
+    )
+
+
 @app.get(
     "/status/{request_id}",
     summary="Get the status of a risk analyzer report",