Adding New Data Fetch Workflows

Relevant source files

The following files were used as context for generating this wiki page:

• src/regime_data_fetch/fetch_workflow.py
• scripts/fetch_regime_engine_v1_data.py
• src/regime_data_fetch/fomc_minutes.py
• src/regime_data_fetch/pmi.py
• src/regime_data_fetch/sf_fed_news_sentiment.py
• src/regime_data_fetch/acquisition_store.py
• AGENTS.md

Adding a new data fetch workflow to the regime-detection engine involves creating a dedicated fetch module, integrating it into the centralized fetch registry, and ensuring it adheres to the project's data acquisition and persistence standards. The system is designed to handle both unattended (automated) and operator-assisted fetches, primarily using a combination of HTTP requests, browser automation fallbacks, and local file imports.

The architecture relies on a FetchModeRegistry that decouples the definition of a data source from the execution logic of the orchestrator. Every workflow must support structured logging, UTC timestamping for persistent state, and row-by-row validation of external inputs.

Sources: scripts/fetch_regime_engine_v1_data.py:74-96, AGENTS.md:54-58, AGENTS.md:73-74

Workflow Architecture and Lifecycle

Data fetch workflows follow a standardized lifecycle: intent announcement, source-specific validation, persistence of raw artifacts, and generation of a normalized output (typically Parquet). The lifecycle is managed by an orchestrator that interacts with an AcquisitionStore to track run metadata and artifact lineage.

Key Lifecycle Components

Component	Responsibility	Relevant Files
Runner Function	Orchestrates the fetch, normalization, and persistence.	fomc_minutes.py, pmi.py
AcquisitionStore	Manages the SQLite database for runs, artifacts, and lineage.	acquisition_store.py
Fetch Report	A JSON summary emitted at the end of every run.	fomc_minutes.py, sf_fed_news_sentiment.py

Integrating with the Fetch Registry

To make a new workflow accessible to the engine, it must be registered in scripts/fetch_regime_engine_v1_data.py. The project uses a FetchModeSpec to define how a workflow is invoked and categorized.

Registry Configuration

Workflows are categorized into two types:

1. UNATTENDED: Automated fetches (e.g., market, macro, sentiment, events) that can be run in parallel.
2. OPERATOR_ASSISTED: Requires manual steps or specific local environments (e.g., EPS workbook imports, browser-based captures).

# Example Registry Entry in scripts/fetch_regime_engine_v1_data.py
FetchModeSpec(
    name="new-datasource", 
    category=UNATTENDED, 
    conservative_concurrent=True, 
    invoke=_invoke_new_datasource_fetch
)

Sources: scripts/fetch_regime_engine_v1_data.py:261-300, TECH_DEBT_AUDIT.md:38-41

Invocation Logic

The invoke callable acts as a bridge between the CLI arguments and the specific run_..._fetch function.

Implementation Standards

New workflows must adhere to specific implementation rules defined in AGENTS.md and evidenced in existing modules:

1. Persistence and Artifact Tracking

Workflows should record raw artifacts before normalization. This allows for historical replay and auditability.

• record_text_artifact: Used for HTML or JSON strings (e.g., FOMC listing pages).
• record_file_artifact: Used for binary or large files (e.g., SF Fed XLSX workbooks).
• record_output: Used for final Parquet files and fetch reports.

Sources: src/regime_data_fetch/fomc_minutes.py:228-237, src/regime_data_fetch/sf_fed_news_sentiment.py:141-150

2. Validation and Error Handling

• Fail-Open for Business Logic: Instrumentation must never block business logic, but validation errors in data must be propagated loudly.
• Source-Specific Validation: Validate data row-by-row. If failure rates exceed 1%, the fetch must stop.
• Deterministic Errors: Propagate errors related to schema or logic immediately. External flakiness (5xx, timeouts) should be caught and logged without killing the entire run.

Sources: AGENTS.md:54-58, AGENTS.md:95-103, src/regime_data_fetch/pmi.py:125-132

3. Data Normalization

Workflows typically produce a Pandas DataFrame that is then written to Parquet. The schema should include standard metadata columns where possible:

• source: A string identifier (e.g., frbsf:daily_news_sentiment).
• source_url: The origin URL of the data.
• date: Standardized datetime64 or ISO-formatted strings.

Sources: src/regime_data_fetch/sf_fed_news_sentiment.py:32-37, src/regime_data_fetch/pmi.py:263-272

Summary

Adding new fetch workflows requires implementing a run_..._fetch function that handles the data lifecycle—acquisition, persistence of raw artifacts via AcquisitionStore, normalization, and final Parquet output. Successful integration is achieved by wrapping this function in an invocation helper and registering it within the FETCH_MODE_REGISTRY in the main fetch script. Adherence to non-negotiable hygiene rules, such as UTC timestamping and structured logging, ensures the stability and auditability of the regime-detection engine's data pipeline.