`recall_guard.dataset.fmp_corpora`

recall_guard.dataset.fmp_corpora

FMP-backed calibration corpus builder for honest-model-ranking.

Implements Requirement 11 of the spec: build the IS/OOS calibration corpora (data/calibration/is_memorized.jsonl label=1 and data/calibration/oos_control.jsonl label=0) from real, dated articles fetched via the Financial Modeling Prep (FMP) news endpoints, plus an update_oos mode that incrementally appends new post-cutoff articles to the OOS corpus only.

The module exposes three public symbols (re-exported from recall_guard.dataset.__init__):

ArticleRecord -- frozen dataclass describing one calibration row.
build_calibration -- one-shot builder driven by the cutoff registry.
update_oos -- incremental refresh of the OOS corpus, never IS.

It also exposes a small fetch_articles helper for direct FMP pagination which the build/update routines compose internally and which is the only location that performs HTTP I/O (requests.get). Tests mock that call.

CLI surface (python -m recall_guard.dataset.fmp_corpora --help):

python -m recall_guard.dataset.fmp_corpora build [--cutoffs PATH] [--out DIR]
                                        [--target N] [--include-stock-news]
python -m recall_guard.dataset.fmp_corpora update [--out DIR] [--since YYYY-MM-DD]

Both subcommands read FMP_API_KEY from the environment (with python-dotenv loading .env when present).

ArticleRecord `dataclass`

One calibration article in canonical in-memory form.

prompt is the concatenated title + body excerpt, capped at _PROMPT_MAX_CHARS characters with surrounding whitespace trimmed. label is 1 for IS-memorized rows (pre-earliest-cutoff) and 0 for OOS rows (post-latest-cutoff).

Source code in recall_guard/dataset/fmp_corpora.py

@dataclass(frozen=True)
class ArticleRecord:
    """One calibration article in canonical in-memory form.

    ``prompt`` is the concatenated title + body excerpt, capped at
    ``_PROMPT_MAX_CHARS`` characters with surrounding whitespace trimmed.
    ``label`` is 1 for IS-memorized rows (pre-earliest-cutoff) and 0 for
    OOS rows (post-latest-cutoff).
    """

    prompt: str
    label: int
    published_at: date
    source: str
    url: str

fetch_articles

fetch_articles(
    endpoint, api_key, from_date, to_date, page, limit
)

Fetch one page of articles from an FMP news endpoint.

Builds the canonical FMP stable/ URL with from, to, page, limit, and apikey query parameters. Raises RuntimeError (with status code + endpoint) on any non-200 response.

Returns the parsed JSON list (assumed to be a list of article dicts).

Source code in recall_guard/dataset/fmp_corpora.py

def fetch_articles(
    endpoint: str,
    api_key: str,
    from_date: date,
    to_date: date,
    page: int,
    limit: int,
) -> list[dict]:
    """Fetch one page of articles from an FMP news endpoint.

    Builds the canonical FMP ``stable/`` URL with ``from``, ``to``,
    ``page``, ``limit``, and ``apikey`` query parameters. Raises
    ``RuntimeError`` (with status code + endpoint) on any non-200 response.

    Returns the parsed JSON list (assumed to be a list of article dicts).
    """
    url = (
        "https://financialmodelingprep.com/stable/"
        f"{endpoint}?from={from_date.isoformat()}&to={to_date.isoformat()}"
        f"&page={page}&limit={limit}&apikey={api_key}"
    )
    response = requests.get(url, timeout=15)
    if response.status_code != 200:
        raise RuntimeError(
            f"FMP endpoint {endpoint!r} returned HTTP {response.status_code}"
        )
    payload = response.json()
    if not isinstance(payload, list):
        # Defensive: design assumes a top-level list.
        return []
    return payload

build_calibration

build_calibration(
    out_dir,
    cutoffs,
    target_per_corpus=100,
    api_key=None,
    endpoints=DEFAULT_ENDPOINTS,
    today=None,
    is_strata=5,
)

Build both calibration corpora from FMP news endpoints.

Filters strictly by publication date (Req 11.2): - IS rows: published BEFORE min(cutoffs.values()); sampled across is_strata equal-width chronological sub-windows of (_EPOCH, earliest_cutoff) so the corpus does not cluster on the cutoff edge (task 1.5). Per-bucket target is target_per_corpus // is_strata; the LAST bucket absorbs any remainder so the totals always sum to target_per_corpus. - OOS rows: published AFTER max(cutoffs.values()) and on/before today. OOS clustering at "now" is acceptable: recent articles are uniformly unseen by every in-registry model, so the OOS sampler keeps a single full window (no stratification). - articles in the gap between earliest and latest cutoff are dropped

Deduplicates by URL exact-match and by sha256(title) (Req 11.3) across ALL sub-windows and ALL endpoints -- a single set per side persists for the whole run, so an article from sub-window 2 cannot reappear under a different bucket in sub-window 3.

Skips articles missing a body or a parseable publishedDate and emits one WARNING per skip (Req 11.4).

Writes out_dir/is_memorized.jsonl and out_dir/oos_control.jsonl as JSONL, one row per line, with the schema::

{"prompt": str, "label": int,
 "metadata": {"published_at": "YYYY-MM-DD",
              "source": str, "url": str}}

Returns (is_path, oos_path).

Source code in recall_guard/dataset/fmp_corpora.py

def build_calibration(
    out_dir: Path | str,
    cutoffs: dict[str, date],
    target_per_corpus: int = 100,
    api_key: str | None = None,
    endpoints: Sequence[str] = DEFAULT_ENDPOINTS,
    today: date | None = None,
    is_strata: int = 5,
) -> tuple[Path, Path]:
    """Build both calibration corpora from FMP news endpoints.

    Filters strictly by publication date (Req 11.2):
      - IS rows: published BEFORE ``min(cutoffs.values())``; sampled across
        ``is_strata`` equal-width chronological sub-windows of
        ``(_EPOCH, earliest_cutoff)`` so the corpus does not cluster on the
        cutoff edge (task 1.5). Per-bucket target is
        ``target_per_corpus // is_strata``; the LAST bucket absorbs any
        remainder so the totals always sum to ``target_per_corpus``.
      - OOS rows: published AFTER ``max(cutoffs.values())`` and on/before
        today. OOS clustering at "now" is acceptable: recent articles are
        uniformly unseen by every in-registry model, so the OOS sampler
        keeps a single full window (no stratification).
      - articles in the gap between earliest and latest cutoff are dropped

    Deduplicates by URL exact-match and by sha256(title) (Req 11.3) across
    ALL sub-windows and ALL endpoints -- a single set per side persists for
    the whole run, so an article from sub-window 2 cannot reappear under a
    different bucket in sub-window 3.

    Skips articles missing a body or a parseable ``publishedDate`` and emits
    one WARNING per skip (Req 11.4).

    Writes ``out_dir/is_memorized.jsonl`` and ``out_dir/oos_control.jsonl``
    as JSONL, one row per line, with the schema::

        {"prompt": str, "label": int,
         "metadata": {"published_at": "YYYY-MM-DD",
                      "source": str, "url": str}}

    Returns ``(is_path, oos_path)``.
    """
    if not cutoffs:
        raise ValueError("cutoffs must be a non-empty mapping of model_id -> date.")
    if is_strata < 1:
        raise ValueError(f"is_strata must be >= 1, got {is_strata}.")

    api_key = _resolve_api_key(api_key)
    today = today or date.today()

    earliest_cutoff = min(cutoffs.values())
    latest_cutoff = max(cutoffs.values())
    if latest_cutoff >= today:
        raise ValueError(
            f"No OOS window available: latest cutoff {latest_cutoff.isoformat()} "
            f">= today {today.isoformat()}."
        )

    out_dir = Path(out_dir)
    is_path = out_dir / "is_memorized.jsonl"
    oos_path = out_dir / "oos_control.jsonl"

    is_records: list[ArticleRecord] = []
    oos_records: list[ArticleRecord] = []
    seen_urls: set[str] = set()
    seen_title_hashes: set[str] = set()

    _collect_is_records(
        endpoints=endpoints, api_key=api_key, today=today,
        earliest_cutoff=earliest_cutoff, latest_cutoff=latest_cutoff,
        is_records=is_records, oos_records=oos_records,
        seen_urls=seen_urls, seen_title_hashes=seen_title_hashes,
        target_per_corpus=target_per_corpus, is_strata=is_strata,
    )
    _collect_oos_records(
        endpoints=endpoints, api_key=api_key, today=today,
        earliest_cutoff=earliest_cutoff, latest_cutoff=latest_cutoff,
        is_records=is_records, oos_records=oos_records,
        seen_urls=seen_urls, seen_title_hashes=seen_title_hashes,
        target_per_corpus=target_per_corpus,
    )
    _warn_shortfall(is_records, oos_records, target_per_corpus, endpoints, is_strata)

    _write_jsonl(is_path, is_records)
    _write_jsonl(oos_path, oos_records)
    return is_path, oos_path

update_oos

update_oos(
    out_dir,
    api_key=None,
    endpoints=DEFAULT_ENDPOINTS,
    today=None,
    since=None,
)

Append new post-cutoff articles to the OOS corpus only (Req 11.5).

Reads the existing out_dir/oos_control.jsonl, derives the latest published_at (or uses since when supplied), fetches articles on or after that date from each endpoint (so late-arriving articles dated on the already-seen max day are still ingestible), dedups against the existing rows (URL + title hash), and appends the new rows in place. Never modifies is_memorized.jsonl.

Raises FileNotFoundError if the OOS file does not exist.

Source code in recall_guard/dataset/fmp_corpora.py

def update_oos(
    out_dir: Path | str,
    api_key: str | None = None,
    endpoints: Sequence[str] = DEFAULT_ENDPOINTS,
    today: date | None = None,
    since: date | None = None,
) -> Path:
    """Append new post-cutoff articles to the OOS corpus only (Req 11.5).

    Reads the existing ``out_dir/oos_control.jsonl``, derives the latest
    ``published_at`` (or uses ``since`` when supplied), fetches articles
    on or after that date from each endpoint (so late-arriving articles
    dated on the already-seen max day are still ingestible), dedups
    against the existing rows (URL + title hash), and appends the new
    rows in place. Never modifies ``is_memorized.jsonl``.

    Raises ``FileNotFoundError`` if the OOS file does not exist.
    """
    out_dir = Path(out_dir)
    oos_path = out_dir / "oos_control.jsonl"
    if not oos_path.exists():
        raise FileNotFoundError(
            f"OOS corpus not found at {oos_path}; run build_calibration first."
        )

    api_key = _resolve_api_key(api_key)
    today = today or date.today()

    existing_rows = _read_existing_oos(oos_path)
    if not existing_rows and since is None:
        raise ValueError(
            f"OOS corpus at {oos_path} is empty; cannot derive since-date. "
            "Pass since=YYYY-MM-DD or rebuild via build_calibration."
        )

    existing_urls, existing_title_hashes, max_published = _index_existing_oos(existing_rows)
    since_date = since if since is not None else max_published
    assert since_date is not None  # established by the empty-file guard above

    # Fetch from the since-date itself, not the day after: an article that
    # became visible late but is dated on the already-seen max day must
    # still have a way in (dedup drops the rows we already hold).
    from_date = since_date
    if from_date > today:
        return oos_path  # nothing to do

    new_records = _fetch_new_oos_records(
        endpoints=endpoints,
        api_key=api_key,
        from_date=from_date,
        today=today,
        since_date=since_date,
        existing_urls=existing_urls,
        existing_title_hashes=existing_title_hashes,
    )
    if new_records:
        _append_jsonl(oos_path, new_records)
    return oos_path

recall_guard.dataset.fmp_corpora