From 30683a4a94cd4e8cfe3f3e83849e9a196d6f9d87 Mon Sep 17 00:00:00 2001 From: PP1 <74917296+pengpengyi92@users.noreply.github.com> Date: Tue, 21 Jul 2026 23:35:22 +0800 Subject: [PATCH 1/2] fix(preprocess): parse shared exchange ticker lists --- contexts/design/flow/news.md | 559 ++++++------ examples/preprocess/02_news_ticker_lists.py | 8 + quantmind/preprocess/news.py | 937 ++++++++++---------- tests/preprocess/test_news.py | 741 ++++++++-------- 4 files changed, 1165 insertions(+), 1080 deletions(-) create mode 100644 examples/preprocess/02_news_ticker_lists.py diff --git a/contexts/design/flow/news.md b/contexts/design/flow/news.md index 98766d5..7c365e1 100644 --- a/contexts/design/flow/news.md +++ b/contexts/design/flow/news.md @@ -1,276 +1,283 @@ -# News Collection Design - -## Quick Summary - -- **Purpose**: Define how QuantMind collects public news and records proof of what each source returned. -- **Read when**: Changing `collect_news`, `NewsWindow`, a news source, the `complete` flag, or news checks. -- **Status**: Current design; the open-source package currently supports PR Newswire. -- **Core rule**: Collection returns documents, failures, and whether the full time window was covered. Other code turns documents into knowledge, stores them, schedules runs, and removes irrelevant news. - -## Contents - -- [Scope and Current Support](#scope-and-current-support) -- [Design Principles](#design-principles) -- [Public API](#public-api) -- [Returned Data](#returned-data) -- [Collection Steps](#collection-steps) -- [Failures and the `complete` Flag](#failures-and-the-complete-flag) -- [Who Owns What](#who-owns-what) -- [How to Verify](#how-to-verify) -- [Out of Scope and When to Add Shared Code](#out-of-scope-and-when-to-add-shared-code) -- [Adding a Public News Source](#adding-a-public-news-source) - -## Scope and Current Support - -This page defines how the open-source package collects public company news. The -first version supports PR Newswire only. It has one collection function, one -time-window input, repeatable HTML cleanup, and visible item failures. - -Its public name follows the -[operation naming rules](../operations/naming.md): collection returns the news -as published plus fetch details. A separate operation turns those documents -into structured knowledge. - -The primary requirement is that any caller can request a complete, one-shot -collection of a past time window. A daily poll is therefore not a separate -operation; it is simply a short window evaluated on a schedule. - -## Design Principles - -1. **Ask for news, not fetch details.** Callers choose a source and time window. - They do not choose RSS, listing pages, how to move between pages, or article - parsing rules. -2. **Return every source row.** QuantMind does not silently remove duplicate - source rows. Repeated rows remain repeated output records, - and may share the same stable ID. -3. **Show partial failures.** The returned batch includes item failures. Invalid - inputs still raise before network work starts. -4. **Hide source implementation details.** PR Newswire may later use a public - mechanism other than listing pages without changing the public function. -5. **List supported sources explicitly.** The first version selects from a - closed source list. It does not expose a provider plugin API or registry. -6. **Keep downstream choices separate.** The calling pipeline owns storage, - duplicate handling, relevance rules, output formats, and scheduling. - -## Public API - -Callers use one entry point: - -```python -from datetime import datetime, timezone - -from quantmind.configs import NewsCollectionCfg, NewsWindow -from quantmind.flows import collect_news - -batch = await collect_news( - NewsWindow( - source="pr-newswire", - start=datetime(2026, 7, 13, tzinfo=timezone.utc), - end=datetime(2026, 7, 14, tzinfo=timezone.utc), - ), - cfg=NewsCollectionCfg(retain_raw_html=False), -) -``` - -`NewsWindow` requires timezone-aware timestamps. Its `[start, end)` interval -includes `start` and excludes `end`. Regular collection and historical runs -use the same call; there are no separate `poll_*`, `backfill_*`, or -`fetch_wire_*` public functions. - -`NewsCollectionCfg` keeps the shared `BaseFlowCfg` fields so it works with the -repository's shared config handling. Its only -collection-specific field is `retain_raw_html`, which controls whether fetched -article HTML bytes remain in the result. It defaults to `False`, which is the -storage-efficient behavior. The article is still fetched, hashed, parsed, and -described by fetch details; only its byte payload is discarded. The collector -does not otherwise use the shared model, tracing, or SDK-run fields. - -### Collection records are not knowledge - -QuantMind keeps collected documents separate from structured knowledge: - -| Operation | Result | Owning package | -|-----------|--------|-----------------| -| `collect_news` | Source documents, fetch details, failures, and whether the full window was scanned | `quantmind.preprocess` | -| future `extract_news_knowledge` | Extracted financial events | `quantmind.knowledge.News` | - -`NewsDocument` is therefore not a `KnowledgeItem`. It carries HTTP details, -raw bytes, parsing output, and collection status that remain useful before any -LLM or business data format is chosen. `knowledge.News` is a structured -financial event with entities, sentiment, financial importance -(`materiality`), source links, and text for embeddings. A pipeline may use both -types, but one cannot replace the other. - -## Returned Data - -`collect_news` returns a `NewsBatch` with four fields. Import these public types -from `quantmind.preprocess`: - -- `documents`: successfully collected `NewsDocument` observations; -- `failures`: lightweight `NewsFailure` records for work that could not be - completed; -- `observed_count`: the number of source rows successfully converted into - in-window observations, before article processing; -- `complete`: whether the listing scan covered the full requested window. - -A `NewsDocument` contains the source name, stable ID, preferred article URL, -title, publisher, publication time, cleaned Markdown, content hash, ticker -hints, and two `NewsArtifact` records: - -- a small record of the public listing row; -- an article record containing fetch details and a content hash. Its - `bytes` field is `None` unless `retain_raw_html=True`. - -`NewsArtifact` records what was fetched. It contains the content hash, -content type, source and resolved URLs, status, headers, and fetch time, with -an optional byte payload. - -Repeated listing rows are not removed. If two rows point to the same release, -the batch contains two observations with the same stable ID. A consumer can -use that ID to update the same database row safely while still recording what -QuantMind observed. - -## Collection Steps - -```text -NewsWindow - -> select the source collector - -> scan public listing pages from newest to oldest - -> keep rows inside [start, end) - -> fetch each linked article - -> convert HTML to Markdown - -> NewsDocument or NewsFailure - -> NewsBatch -``` - -PR Newswire discovery is based on its public news-release listing rather than -the latest-items RSS snapshot. Pages are read newest to oldest until an -observation strictly older than the requested start is seen. The strict -boundary matters because several rows with the same minute-level timestamp may -span two pages. A caller can therefore rerun a past-day request without a -previously saved cursor. - -PR Newswire exposes listing timestamps at minute precision. Scheduled callers -should therefore use minute-aligned window bounds, as the live end-to-end check -does. - -RSS remains an internal parser and a live check. It is not a public -`NewsInput`: choosing a feed is an implementation detail, and a limited feed -snapshot cannot prove that it covered a complete time window. - -The HTTP layer limits retries, waits between attempts, honors `Retry-After`, -and limits requests per host. PR Newswire-specific URL construction and -listing HTML parsing stay in the PR Newswire source module. - -`NewsWindow.source` lists every supported source. `collect_news` has an -explicit branch for each one, so the type checker catches a source name added -without a matching collector. - -## Failures and the `complete` Flag - -Configuration errors raise immediately. Examples include a naive timestamp, -an empty source, an unsupported source, or `end <= start`. - -After collection begins, one item failure does not stop independent items. -Each `NewsFailure` records the source, failed step, URL, optional item ID, error -category, and message. - -`complete=False` when any of the following is true: - -- a listing page could not be fetched or parsed; -- the listing scan stopped before crossing the window start. - -Article failures stay in `failures` but do not change whether all listing rows -were found. A caller can distinguish "the full time window was scanned" from -"every found article was processed." It can store successful records and retry -failed articles separately. An empty batch is complete only when the listing -scan crossed the requested start. - -## Who Owns What - -QuantMind owns: - -- scanning public listings and fetching articles; -- repeatable HTML cleanup; -- stable IDs and content hashes; -- honest batch counts, failure records, and completeness. - -The consuming production pipeline owns: - -- its schedule and when its GitHub Action runs; -- database storage, saved progress, and safe repeated writes; -- rules for removing duplicates; -- rules that remove irrelevant company news; -- later data formats, added fields, and database writes; -- shared batch callbacks, metrics, and monitoring. - -This split lets a separate ingestion job request the last day in one call, -apply its own rules, and write its own data format without adding those choices -to the open-source library. - -## How to Verify - -Use separate offline and live checks. - -### Offline repository checks - -`bash scripts/verify.sh` runs repeatable unit tests, linting, typing, and -coverage. News tests use saved HTML/RSS files and mocked HTTP responses. -They cover window validation, time edges, multi-page listings, duplicate -observations, raw-byte retention, retries, partial failures, and completeness. - -### Live end-to-end news check - -`python scripts/verify_news_e2e.py` is the news live-network check. It performs -three limited checks: - -1. fetch and parse the official PR Newswire RSS feed; -2. scan PR Newswire listing rows for the preceding 24 hours and prove that the - scan crossed the window start; -3. fetch up to 25 unique articles and find the first supported exchange-coded - symbol in every group that has one (100% recall). - -The ticker check uses a separate comparison regex rather than the production -extractor. It accepts plain, Markdown-link, and emphasis-wrapped symbols. It -ignores later symbols in a multi-symbol list and exchanges that the extractor -does not support. The script prints PASS/FAIL plus short listing, article, -failure, and recall summaries. It fails when RSS or listing data is invalid, -no sampled article parses, or supported symbols fall below 100% recall. A -parsed sample with no supported symbols reports `SKIP` and still passes. - -The `news` job in `.github/workflows/e2e.yml` runs this check once daily, when -started manually, and on pull requests that change its direct dependencies. -The required `.github/workflows/ci.yml` workflow remains network-free so local -development and unit tests remain repeatable. - -## Out of Scope and When to Add Shared Code - -The first version does not include GlobeNewswire, Business Wire, authenticated -feeds, continuous cursors, storage, duplicate removal, financial importance -scoring, or a generic batch-operation base class. - -When a second source is implemented, compare its real behavior with PR -Newswire first. Add a shared collector interface only for behavior that both -working collectors actually share. Keep the public -`collect_news(NewsWindow, *, cfg)` call unchanged. - -## Adding a Public News Source - -A coding agent adding a second source follows this checklist: - -1. Add the source name to `NewsWindow.source`. -2. Add one private `quantmind/preprocess/.py` collector. -3. Add one explicit branch to `collect_news`; the type checker must pass. -4. Add saved-input tests for success, time edges, duplicate observations, - partial failures, and the `complete` flag. -5. Add a test proving that only the selected collector is called. -6. Update the supported-source table in `docs/README.md`, this design, and the - focused example if its common path changes. -7. Add or extend a component-specific live verifier and its named job in the - existing `.github/workflows/e2e.yml` when the integration depends on a - public network endpoint. Add its command to `docs/README.md`; do not add the - command to root agent guidance. - -Only after two real collectors share behavior should they use a common Python -`Protocol`. Never add a source only to the input `Literal`; the type checker is -expected to reject that incomplete change. +# News Collection Design + +## Quick Summary + +- **Purpose**: Define how QuantMind collects public news and records proof of what each source returned. +- **Read when**: Changing `collect_news`, `NewsWindow`, a news source, the `complete` flag, or news checks. +- **Status**: Current design; the open-source package currently supports PR Newswire. +- **Core rule**: Collection returns documents, failures, and whether the full time window was covered. Other code turns documents into knowledge, stores them, schedules runs, and removes irrelevant news. + +## Contents + +- [Scope and Current Support](#scope-and-current-support) +- [Design Principles](#design-principles) +- [Public API](#public-api) +- [Returned Data](#returned-data) +- [Collection Steps](#collection-steps) +- [Failures and the `complete` Flag](#failures-and-the-complete-flag) +- [Who Owns What](#who-owns-what) +- [How to Verify](#how-to-verify) +- [Out of Scope and When to Add Shared Code](#out-of-scope-and-when-to-add-shared-code) +- [Adding a Public News Source](#adding-a-public-news-source) + +## Scope and Current Support + +This page defines how the open-source package collects public company news. The +first version supports PR Newswire only. It has one collection function, one +time-window input, repeatable HTML cleanup, and visible item failures. + +Its public name follows the +[operation naming rules](../operations/naming.md): collection returns the news +as published plus fetch details. A separate operation turns those documents +into structured knowledge. + +The primary requirement is that any caller can request a complete, one-shot +collection of a past time window. A daily poll is therefore not a separate +operation; it is simply a short window evaluated on a schedule. + +## Design Principles + +1. **Ask for news, not fetch details.** Callers choose a source and time window. + They do not choose RSS, listing pages, how to move between pages, or article + parsing rules. +2. **Return every source row.** QuantMind does not silently remove duplicate + source rows. Repeated rows remain repeated output records, + and may share the same stable ID. +3. **Show partial failures.** The returned batch includes item failures. Invalid + inputs still raise before network work starts. +4. **Hide source implementation details.** PR Newswire may later use a public + mechanism other than listing pages without changing the public function. +5. **List supported sources explicitly.** The first version selects from a + closed source list. It does not expose a provider plugin API or registry. +6. **Keep downstream choices separate.** The calling pipeline owns storage, + duplicate handling, relevance rules, output formats, and scheduling. + +## Public API + +Callers use one entry point: + +```python +from datetime import datetime, timezone + +from quantmind.configs import NewsCollectionCfg, NewsWindow +from quantmind.flows import collect_news + +batch = await collect_news( + NewsWindow( + source="pr-newswire", + start=datetime(2026, 7, 13, tzinfo=timezone.utc), + end=datetime(2026, 7, 14, tzinfo=timezone.utc), + ), + cfg=NewsCollectionCfg(retain_raw_html=False), +) +``` + +`NewsWindow` requires timezone-aware timestamps. Its `[start, end)` interval +includes `start` and excludes `end`. Regular collection and historical runs +use the same call; there are no separate `poll_*`, `backfill_*`, or +`fetch_wire_*` public functions. + +`NewsCollectionCfg` keeps the shared `BaseFlowCfg` fields so it works with the +repository's shared config handling. Its only +collection-specific field is `retain_raw_html`, which controls whether fetched +article HTML bytes remain in the result. It defaults to `False`, which is the +storage-efficient behavior. The article is still fetched, hashed, parsed, and +described by fetch details; only its byte payload is discarded. The collector +does not otherwise use the shared model, tracing, or SDK-run fields. + +### Collection records are not knowledge + +QuantMind keeps collected documents separate from structured knowledge: + +| Operation | Result | Owning package | +|-----------|--------|-----------------| +| `collect_news` | Source documents, fetch details, failures, and whether the full window was scanned | `quantmind.preprocess` | +| future `extract_news_knowledge` | Extracted financial events | `quantmind.knowledge.News` | + +`NewsDocument` is therefore not a `KnowledgeItem`. It carries HTTP details, +raw bytes, parsing output, and collection status that remain useful before any +LLM or business data format is chosen. `knowledge.News` is a structured +financial event with entities, sentiment, financial importance +(`materiality`), source links, and text for embeddings. A pipeline may use both +types, but one cannot replace the other. + +## Returned Data + +`collect_news` returns a `NewsBatch` with four fields. Import these public types +from `quantmind.preprocess`: + +- `documents`: successfully collected `NewsDocument` observations; +- `failures`: lightweight `NewsFailure` records for work that could not be + completed; +- `observed_count`: the number of source rows successfully converted into + in-window observations, before article processing; +- `complete`: whether the listing scan covered the full requested window. + +A `NewsDocument` contains the source name, stable ID, preferred article URL, +title, publisher, publication time, cleaned Markdown, content hash, ticker +hints, and two `NewsArtifact` records: + +- a small record of the public listing row; +- an article record containing fetch details and a content hash. Its + `bytes` field is `None` unless `retain_raw_html=True`. + +`NewsArtifact` records what was fetched. It contains the content hash, +content type, source and resolved URLs, status, headers, and fetch time, with +an optional byte payload. + +Repeated listing rows are not removed. If two rows point to the same release, +the batch contains two observations with the same stable ID. A consumer can +use that ID to update the same database row safely while still recording what +QuantMind observed. + +## Collection Steps + +```text +NewsWindow + -> select the source collector + -> scan public listing pages from newest to oldest + -> keep rows inside [start, end) + -> fetch each linked article + -> convert HTML to Markdown + -> NewsDocument or NewsFailure + -> NewsBatch +``` + +PR Newswire discovery is based on its public news-release listing rather than +the latest-items RSS snapshot. Pages are read newest to oldest until an +observation strictly older than the requested start is seen. The strict +boundary matters because several rows with the same minute-level timestamp may +span two pages. A caller can therefore rerun a past-day request without a +previously saved cursor. + +PR Newswire exposes listing timestamps at minute precision. Scheduled callers +should therefore use minute-aligned window bounds, as the live end-to-end check +does. + +RSS remains an internal parser and a live check. It is not a public +`NewsInput`: choosing a feed is an implementation detail, and a limited feed +snapshot cannot prove that it covered a complete time window. + +The HTTP layer limits retries, waits between attempts, honors `Retry-After`, +and limits requests per host. PR Newswire-specific URL construction and +listing HTML parsing stay in the PR Newswire source module. + +`NewsWindow.source` lists every supported source. `collect_news` has an +explicit branch for each one, so the type checker catches a source name added +without a matching collector. + +## Failures and the `complete` Flag + +Configuration errors raise immediately. Examples include a naive timestamp, +an empty source, an unsupported source, or `end <= start`. + +After collection begins, one item failure does not stop independent items. +Each `NewsFailure` records the source, failed step, URL, optional item ID, error +category, and message. + +`complete=False` when any of the following is true: + +- a listing page could not be fetched or parsed; +- the listing scan stopped before crossing the window start. + +Article failures stay in `failures` but do not change whether all listing rows +were found. A caller can distinguish "the full time window was scanned" from +"every found article was processed." It can store successful records and retry +failed articles separately. An empty batch is complete only when the listing +scan crossed the requested start. + +## Who Owns What + +QuantMind owns: + +- scanning public listings and fetching articles; +- repeatable HTML cleanup; +- stable IDs and content hashes; +- honest batch counts, failure records, and completeness. + +The consuming production pipeline owns: + +- its schedule and when its GitHub Action runs; +- database storage, saved progress, and safe repeated writes; +- rules for removing duplicates; +- rules that remove irrelevant company news; +- later data formats, added fields, and database writes; +- shared batch callbacks, metrics, and monitoring. + +This split lets a separate ingestion job request the last day in one call, +apply its own rules, and write its own data format without adding those choices +to the open-source library. + +## How to Verify + +Use separate offline and live checks. + +### Offline repository checks + +`bash scripts/verify.sh` runs repeatable unit tests, linting, typing, and +coverage. News tests use saved HTML/RSS files and mocked HTTP responses. +They cover window validation, time edges, multi-page listings, duplicate +observations, raw-byte retention, retries, partial failures, and completeness. + +### Live end-to-end news check + +`python scripts/verify_news_e2e.py` is the news live-network check. It performs +three limited checks: + +1. fetch and parse the official PR Newswire RSS feed; +2. scan PR Newswire listing rows for the preceding 24 hours and prove that the + scan crossed the window start; +3. fetch up to 25 unique articles and find the first supported exchange-coded + symbol in every group that has one (100% recall). + +Ticker-hint extraction supports one exchange-qualified symbol and parenthesized +comma-separated symbol lists that share a supported exchange prefix. A shared +group ends at a semicolon, conjunction, or closing parenthesis. Additional list +members record a reconstructed `EXCHANGE: SYMBOL` raw value; unsupported +exchanges remain outside the whitelist policy. + +The ticker check uses a separate comparison regex rather than the production +extractor. It accepts plain, Markdown-link, and emphasis-wrapped symbols. It +recognizes comma-separated symbols that share a supported exchange prefix and +ignores exchanges that the extractor does not support. The script prints +PASS/FAIL plus short listing, article, +failure, and recall summaries. It fails when RSS or listing data is invalid, +no sampled article parses, or supported symbols fall below 100% recall. A +parsed sample with no supported symbols reports `SKIP` and still passes. + +The `news` job in `.github/workflows/e2e.yml` runs this check once daily, when +started manually, and on pull requests that change its direct dependencies. +The required `.github/workflows/ci.yml` workflow remains network-free so local +development and unit tests remain repeatable. + +## Out of Scope and When to Add Shared Code + +The first version does not include GlobeNewswire, Business Wire, authenticated +feeds, continuous cursors, storage, duplicate removal, financial importance +scoring, or a generic batch-operation base class. + +When a second source is implemented, compare its real behavior with PR +Newswire first. Add a shared collector interface only for behavior that both +working collectors actually share. Keep the public +`collect_news(NewsWindow, *, cfg)` call unchanged. + +## Adding a Public News Source + +A coding agent adding a second source follows this checklist: + +1. Add the source name to `NewsWindow.source`. +2. Add one private `quantmind/preprocess/.py` collector. +3. Add one explicit branch to `collect_news`; the type checker must pass. +4. Add saved-input tests for success, time edges, duplicate observations, + partial failures, and the `complete` flag. +5. Add a test proving that only the selected collector is called. +6. Update the supported-source table in `docs/README.md`, this design, and the + focused example if its common path changes. +7. Add or extend a component-specific live verifier and its named job in the + existing `.github/workflows/e2e.yml` when the integration depends on a + public network endpoint. Add its command to `docs/README.md`; do not add the + command to root agent guidance. + +Only after two real collectors share behavior should they use a common Python +`Protocol`. Never add a source only to the input `Literal`; the type checker is +expected to reject that incomplete change. diff --git a/examples/preprocess/02_news_ticker_lists.py b/examples/preprocess/02_news_ticker_lists.py new file mode 100644 index 0000000..0700114 --- /dev/null +++ b/examples/preprocess/02_news_ticker_lists.py @@ -0,0 +1,8 @@ +"""Extract ticker hints from a shared exchange-prefix list.""" + +from quantmind.preprocess import extract_exchange_ticker_hints + +text = "Example issuer (NYSE: EVEX, EVEXW; B3: EVEB31) announced results." + +for hint in extract_exchange_ticker_hints(text): + print(hint) diff --git a/quantmind/preprocess/news.py b/quantmind/preprocess/news.py index 12e7706..fffcc65 100644 --- a/quantmind/preprocess/news.py +++ b/quantmind/preprocess/news.py @@ -1,453 +1,484 @@ -"""Source-agnostic news preprocessing and candidate normalization.""" - -import hashlib -import re -from dataclasses import dataclass, field -from datetime import datetime -from typing import Literal -from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit - -from quantmind.preprocess._news_types import NewsTickerHint -from quantmind.preprocess.clean import ( - collapse_whitespace, - dedupe_lines, - normalize_unicode, -) -from quantmind.preprocess.fetch._types import Fetched -from quantmind.preprocess.fetch.http import HttpFetcher, fetch_url -from quantmind.preprocess.fetch.rss import FeedItem -from quantmind.preprocess.format.html import html_to_markdown -from quantmind.preprocess.time import to_utc - -NewsSourceType = Literal[ - "company_8k", - "press_release", - "publisher_news", - "regulatory_news", -] -BodySource = Literal["feed", "article"] - -_SOURCE_PREFIX: dict[str, str] = { - "company_8k": "sec", - "press_release": "wire", - "publisher_news": "publisher", - "regulatory_news": "regulatory", -} - -_TRACKING_QUERY_PARAMS = { - "fbclid", - "feedref", - "gclid", - "mc_cid", - "mc_eid", - "spm", -} - -_EXCHANGE_TICKER_RE = re.compile( - r"(?:\(|\b)" - r"(NASDAQ|NYSE(?:\s+American|\s+MKT|\s+Arca)?|AMEX|OTCQX|OTCQB|OTC|CBOE)" - r"\s*:\s*" - r"([A-Z][A-Z0-9.-]{0,9})" - r"(?:\)|\b)", - re.IGNORECASE, -) -_MARKDOWN_LINK_RE = re.compile(r"\[([^\]\n]+)\]\([^\)\n]*\)") -_MARKDOWN_EMPHASIS_RE = re.compile( - r"(? RawNewsDocument: - """Fetch one news/press-release URL and extract plain markdown text. - - Args: - url: Article or press-release URL. - source_type: Normalised source family. - title: Optional title supplied by a feed or caller. - publisher: Optional source publisher. - published_at: Optional publication timestamp. - payload_id: Optional upstream feed/item id. - timeout: Per-request timeout in seconds. - max_bytes: Hard ceiling on response body size. - fetcher: Optional shared HTTP fetcher and host-rate state. - - Returns: - Raw document with extracted markdown text and caller metadata. - - Raises: - ValueError: If the content type cannot be converted to text. - httpx.HTTPError: For network / status / timeout failures. - """ - if fetcher is None: - fetched = await fetch_url(url, timeout=timeout, max_bytes=max_bytes) - else: - fetched = await fetcher.fetch_url( - url, - timeout=timeout, - max_bytes=max_bytes, - ) - return await news_document_from_fetched( - fetched, - source_type=source_type, - title=title, - publisher=publisher, - published_at=published_at, - payload_id=payload_id, - ) - - -async def news_document_from_fetched( - fetched: Fetched, - *, - source_type: NewsSourceType = "press_release", - title: str | None = None, - publisher: str | None = None, - published_at: datetime | None = None, - payload_id: str | None = None, -) -> RawNewsDocument: - """Extract a raw news document from already-fetched evidence.""" - body_text = await _text_from_fetched(fetched) - return RawNewsDocument( - body_text=body_text, - source_type=source_type, - source_url=fetched.resolved_url or fetched.source_url, - title=title, - publisher=publisher, - published_at=published_at, - payload_id=payload_id, - metadata={ - "body_source": "article", - "content_type": fetched.content_type, - }, - ) - - -async def preprocess_news_url( - url: str, - *, - source_type: NewsSourceType = "press_release", - title: str | None = None, - publisher: str | None = None, - published_at: datetime | None = None, - payload_id: str | None = None, - timeout: float = 30.0, - max_bytes: int = 10_000_000, -) -> NewsCandidate: - """Fetch and normalise one news/press-release URL.""" - raw = await fetch_news_document( - url, - source_type=source_type, - title=title, - publisher=publisher, - published_at=published_at, - payload_id=payload_id, - timeout=timeout, - max_bytes=max_bytes, - ) - return preprocess_news_document(raw) - - -async def feed_item_to_news_document( - item: FeedItem, - *, - source_type: NewsSourceType = "press_release", - publisher: str | None = None, - body_source: BodySource = "feed", - fetcher: HttpFetcher | None = None, -) -> RawNewsDocument: - """Convert one parsed RSS/Atom item into a raw news document. - - ``body_source="feed"`` trusts the RSS/Atom body. ``"article"`` always - follows the item URL, even when the feed contains a non-empty teaser. - """ - if body_source == "article": - if not item.url: - raise ValueError("article body_source requires an item URL") - raw = await fetch_news_document( - item.url, - source_type=source_type, - title=item.title, - publisher=publisher, - published_at=item.published_at, - payload_id=item.id, - fetcher=fetcher, - ) - return RawNewsDocument( - body_text=raw.body_text, - source_type=raw.source_type, - source_url=raw.source_url, - title=raw.title, - publisher=raw.publisher, - published_at=raw.published_at, - payload_id=raw.payload_id, - metadata={ - **raw.metadata, - **{ - key: value - for key, value in { - "source_feed_url": item.source_feed_url, - **item.raw, - }.items() - if value - }, - }, - ) - - body_text = await _html_or_text_to_markdown( - item.content_html or item.summary_html or "" - ) - return RawNewsDocument( - body_text=body_text, - source_type=source_type, - source_url=item.url, - title=item.title, - publisher=publisher, - published_at=item.published_at, - payload_id=item.id, - metadata={ - key: value - for key, value in { - "body_source": "feed", - "source_feed_url": item.source_feed_url, - **item.raw, - }.items() - if value - }, - ) - - -async def preprocess_feed_item( - item: FeedItem, - *, - source_type: NewsSourceType = "press_release", - publisher: str | None = None, - body_source: BodySource = "feed", - fetcher: HttpFetcher | None = None, -) -> NewsCandidate: - """Convert and normalise one parsed RSS/Atom item.""" - raw = await feed_item_to_news_document( - item, - source_type=source_type, - publisher=publisher, - body_source=body_source, - fetcher=fetcher, - ) - return preprocess_news_document(raw) - - -def preprocess_news_document(raw: RawNewsDocument) -> NewsCandidate: - """Normalise a raw news document into the shared candidate contract. - - The output carries stable source identity, cleaned body text, a content - hash, provenance, timestamp, and deterministic ticker hints. - """ - body_text = normalize_news_text(raw.body_text) - if not body_text: - raise ValueError("news document body is empty after preprocessing") - - source_url = ( - canonicalize_source_url(raw.source_url) if raw.source_url else None - ) - title = normalize_news_text(raw.title or "") or None - published_at = to_utc(raw.published_at) if raw.published_at else None - ticker_scan = "\n".join(part for part in (title, body_text) if part) - - return NewsCandidate( - body_text=body_text, - content_hash=news_content_hash(body_text), - source_type=raw.source_type, - identity=build_news_identity( - source_type=raw.source_type, - source_url=source_url, - payload_id=raw.payload_id, - ), - source_url=source_url, - title=title, - publisher=raw.publisher, - published_at=published_at, - ticker_hints=extract_exchange_ticker_hints(ticker_scan), - metadata=dict(raw.metadata), - ) - - -def normalize_news_text(text: str) -> str: - """Apply QuantMind's canonical text cleanup order for news bodies.""" - normalized = normalize_unicode(text) - normalized = _EMAIL_PROTECTION_LINK_RE.sub("[email protected]", normalized) - return collapse_whitespace(dedupe_lines(normalized)) - - -def news_content_hash(text: str) -> str: - """Return the sha256 hash for an already-normalised news body.""" - return hashlib.sha256(text.encode("utf-8")).hexdigest() - - -def build_news_identity( - *, - source_type: NewsSourceType, - source_url: str | None = None, - payload_id: str | None = None, -) -> str: - """Build a deterministic source-document identity. - - Press-release/wire rows use the payload id when available and otherwise - fall back to the canonical source URL. The identity is hashed so callers do - not accidentally depend on upstream id formatting. - """ - identity = (payload_id or "").strip() - if not identity and source_url: - identity = canonicalize_source_url(source_url) - if not identity: - raise ValueError("news identity requires payload_id or source_url") - prefix = _SOURCE_PREFIX[source_type] - digest = hashlib.sha256(identity.encode("utf-8")).hexdigest() - return f"{prefix}:{digest}" - - -def build_sec_news_identity( - *, - accession_number: str, - section_key: str, -) -> str: - """Build a stable 8-K EX-99.x news identity.""" - accession = accession_number.strip() - section = section_key.strip().lower() - if not accession or not section: - raise ValueError("SEC news identity requires accession and section") - return f"sec:{accession}:{section}" - - -def canonicalize_source_url(url: str) -> str: - """Normalise source URLs for stable source identity.""" - parsed = urlsplit(url.strip()) - query = [ - (key, value) - for key, value in parse_qsl(parsed.query, keep_blank_values=True) - if not key.lower().startswith("utm_") - and key.lower() not in _TRACKING_QUERY_PARAMS - ] - path = parsed.path or "/" - if path != "/": - path = path.rstrip("/") - return urlunsplit( - ( - parsed.scheme.lower(), - parsed.netloc.lower(), - path, - urlencode(sorted(query), doseq=True), - "", - ) - ) - - -def extract_exchange_ticker_hints(text: str) -> tuple[NewsTickerHint, ...]: - """Extract exchange-qualified ticker mentions from PR-style text. - - Examples matched include ``(NASDAQ: NVDA)`` and ``NYSE: IBM``. The result is - only a hint; downstream instrument resolution should still validate it. - Markdown link and emphasis decoration is removed from a scan-only copy so - stored news text and its content hash retain their original representation. - """ - scan_text = _MARKDOWN_LINK_RE.sub(r"\1", text) - scan_text = _MARKDOWN_EMPHASIS_RE.sub(r"\1", scan_text) - hints: list[NewsTickerHint] = [] - seen: set[tuple[str, str | None]] = set() - for match in _EXCHANGE_TICKER_RE.finditer(scan_text): - raw_exchange = " ".join(match.group(1).upper().split()) - exchange = _EXCHANGE_NAMES.get(raw_exchange, raw_exchange) - symbol = match.group(2).upper() - key = (symbol, exchange) - if key in seen: - continue - seen.add(key) - hints.append( - NewsTickerHint( - symbol=symbol, - exchange=exchange, - raw=match.group(0).strip(), - ) - ) - return tuple(hints) - - -async def _text_from_fetched(raw: Fetched) -> str: - ct = (raw.content_type or "").lower() - text = raw.bytes.decode("utf-8", errors="replace") - if ct.startswith("text/html") or ct.startswith("application/xhtml+xml"): - return await html_to_markdown(text) - if ( - ct.startswith("text/plain") - or ct.startswith("text/markdown") - or ct.startswith("application/xml") - or ct.startswith("text/xml") - ): - return text - raise ValueError(f"Unsupported content-type for news input: {ct!r}") - - -async def _html_or_text_to_markdown(value: str) -> str: - text = value.strip() - if not text: - return "" - if "<" not in text or ">" not in text: - return text - markdown = await html_to_markdown(text) - return markdown or text +"""Source-agnostic news preprocessing and candidate normalization.""" + +import hashlib +import re +from dataclasses import dataclass, field +from datetime import datetime +from typing import Literal +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + +from quantmind.preprocess._news_types import NewsTickerHint +from quantmind.preprocess.clean import ( + collapse_whitespace, + dedupe_lines, + normalize_unicode, +) +from quantmind.preprocess.fetch._types import Fetched +from quantmind.preprocess.fetch.http import HttpFetcher, fetch_url +from quantmind.preprocess.fetch.rss import FeedItem +from quantmind.preprocess.format.html import html_to_markdown +from quantmind.preprocess.time import to_utc + +NewsSourceType = Literal[ + "company_8k", + "press_release", + "publisher_news", + "regulatory_news", +] +BodySource = Literal["feed", "article"] + +_SOURCE_PREFIX: dict[str, str] = { + "company_8k": "sec", + "press_release": "wire", + "publisher_news": "publisher", + "regulatory_news": "regulatory", +} + +_TRACKING_QUERY_PARAMS = { + "fbclid", + "feedref", + "gclid", + "mc_cid", + "mc_eid", + "spm", +} + +_EXCHANGE_TICKER_RE = re.compile( + r"(?:\(|\b)" + r"(NASDAQ|NYSE(?:\s+American|\s+MKT|\s+Arca)?|AMEX|OTCQX|OTCQB|OTC|CBOE)" + r"\s*:\s*" + r"([A-Z][A-Z0-9.-]{0,9})" + r"(?:\)|\b)", + re.IGNORECASE, +) +_MARKDOWN_LINK_RE = re.compile(r"\[([^\]\n]+)\]\([^\)\n]*\)") +_MARKDOWN_EMPHASIS_RE = re.compile( + r"(? RawNewsDocument: + """Fetch one news/press-release URL and extract plain markdown text. + + Args: + url: Article or press-release URL. + source_type: Normalised source family. + title: Optional title supplied by a feed or caller. + publisher: Optional source publisher. + published_at: Optional publication timestamp. + payload_id: Optional upstream feed/item id. + timeout: Per-request timeout in seconds. + max_bytes: Hard ceiling on response body size. + fetcher: Optional shared HTTP fetcher and host-rate state. + + Returns: + Raw document with extracted markdown text and caller metadata. + + Raises: + ValueError: If the content type cannot be converted to text. + httpx.HTTPError: For network / status / timeout failures. + """ + if fetcher is None: + fetched = await fetch_url(url, timeout=timeout, max_bytes=max_bytes) + else: + fetched = await fetcher.fetch_url( + url, + timeout=timeout, + max_bytes=max_bytes, + ) + return await news_document_from_fetched( + fetched, + source_type=source_type, + title=title, + publisher=publisher, + published_at=published_at, + payload_id=payload_id, + ) + + +async def news_document_from_fetched( + fetched: Fetched, + *, + source_type: NewsSourceType = "press_release", + title: str | None = None, + publisher: str | None = None, + published_at: datetime | None = None, + payload_id: str | None = None, +) -> RawNewsDocument: + """Extract a raw news document from already-fetched evidence.""" + body_text = await _text_from_fetched(fetched) + return RawNewsDocument( + body_text=body_text, + source_type=source_type, + source_url=fetched.resolved_url or fetched.source_url, + title=title, + publisher=publisher, + published_at=published_at, + payload_id=payload_id, + metadata={ + "body_source": "article", + "content_type": fetched.content_type, + }, + ) + + +async def preprocess_news_url( + url: str, + *, + source_type: NewsSourceType = "press_release", + title: str | None = None, + publisher: str | None = None, + published_at: datetime | None = None, + payload_id: str | None = None, + timeout: float = 30.0, + max_bytes: int = 10_000_000, +) -> NewsCandidate: + """Fetch and normalise one news/press-release URL.""" + raw = await fetch_news_document( + url, + source_type=source_type, + title=title, + publisher=publisher, + published_at=published_at, + payload_id=payload_id, + timeout=timeout, + max_bytes=max_bytes, + ) + return preprocess_news_document(raw) + + +async def feed_item_to_news_document( + item: FeedItem, + *, + source_type: NewsSourceType = "press_release", + publisher: str | None = None, + body_source: BodySource = "feed", + fetcher: HttpFetcher | None = None, +) -> RawNewsDocument: + """Convert one parsed RSS/Atom item into a raw news document. + + ``body_source="feed"`` trusts the RSS/Atom body. ``"article"`` always + follows the item URL, even when the feed contains a non-empty teaser. + """ + if body_source == "article": + if not item.url: + raise ValueError("article body_source requires an item URL") + raw = await fetch_news_document( + item.url, + source_type=source_type, + title=item.title, + publisher=publisher, + published_at=item.published_at, + payload_id=item.id, + fetcher=fetcher, + ) + return RawNewsDocument( + body_text=raw.body_text, + source_type=raw.source_type, + source_url=raw.source_url, + title=raw.title, + publisher=raw.publisher, + published_at=raw.published_at, + payload_id=raw.payload_id, + metadata={ + **raw.metadata, + **{ + key: value + for key, value in { + "source_feed_url": item.source_feed_url, + **item.raw, + }.items() + if value + }, + }, + ) + + body_text = await _html_or_text_to_markdown( + item.content_html or item.summary_html or "" + ) + return RawNewsDocument( + body_text=body_text, + source_type=source_type, + source_url=item.url, + title=item.title, + publisher=publisher, + published_at=item.published_at, + payload_id=item.id, + metadata={ + key: value + for key, value in { + "body_source": "feed", + "source_feed_url": item.source_feed_url, + **item.raw, + }.items() + if value + }, + ) + + +async def preprocess_feed_item( + item: FeedItem, + *, + source_type: NewsSourceType = "press_release", + publisher: str | None = None, + body_source: BodySource = "feed", + fetcher: HttpFetcher | None = None, +) -> NewsCandidate: + """Convert and normalise one parsed RSS/Atom item.""" + raw = await feed_item_to_news_document( + item, + source_type=source_type, + publisher=publisher, + body_source=body_source, + fetcher=fetcher, + ) + return preprocess_news_document(raw) + + +def preprocess_news_document(raw: RawNewsDocument) -> NewsCandidate: + """Normalise a raw news document into the shared candidate contract. + + The output carries stable source identity, cleaned body text, a content + hash, provenance, timestamp, and deterministic ticker hints. + """ + body_text = normalize_news_text(raw.body_text) + if not body_text: + raise ValueError("news document body is empty after preprocessing") + + source_url = ( + canonicalize_source_url(raw.source_url) if raw.source_url else None + ) + title = normalize_news_text(raw.title or "") or None + published_at = to_utc(raw.published_at) if raw.published_at else None + ticker_scan = "\n".join(part for part in (title, body_text) if part) + + return NewsCandidate( + body_text=body_text, + content_hash=news_content_hash(body_text), + source_type=raw.source_type, + identity=build_news_identity( + source_type=raw.source_type, + source_url=source_url, + payload_id=raw.payload_id, + ), + source_url=source_url, + title=title, + publisher=raw.publisher, + published_at=published_at, + ticker_hints=extract_exchange_ticker_hints(ticker_scan), + metadata=dict(raw.metadata), + ) + + +def normalize_news_text(text: str) -> str: + """Apply QuantMind's canonical text cleanup order for news bodies.""" + normalized = normalize_unicode(text) + normalized = _EMAIL_PROTECTION_LINK_RE.sub("[email protected]", normalized) + return collapse_whitespace(dedupe_lines(normalized)) + + +def news_content_hash(text: str) -> str: + """Return the sha256 hash for an already-normalised news body.""" + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def build_news_identity( + *, + source_type: NewsSourceType, + source_url: str | None = None, + payload_id: str | None = None, +) -> str: + """Build a deterministic source-document identity. + + Press-release/wire rows use the payload id when available and otherwise + fall back to the canonical source URL. The identity is hashed so callers do + not accidentally depend on upstream id formatting. + """ + identity = (payload_id or "").strip() + if not identity and source_url: + identity = canonicalize_source_url(source_url) + if not identity: + raise ValueError("news identity requires payload_id or source_url") + prefix = _SOURCE_PREFIX[source_type] + digest = hashlib.sha256(identity.encode("utf-8")).hexdigest() + return f"{prefix}:{digest}" + + +def build_sec_news_identity( + *, + accession_number: str, + section_key: str, +) -> str: + """Build a stable 8-K EX-99.x news identity.""" + accession = accession_number.strip() + section = section_key.strip().lower() + if not accession or not section: + raise ValueError("SEC news identity requires accession and section") + return f"sec:{accession}:{section}" + + +def canonicalize_source_url(url: str) -> str: + """Normalise source URLs for stable source identity.""" + parsed = urlsplit(url.strip()) + query = [ + (key, value) + for key, value in parse_qsl(parsed.query, keep_blank_values=True) + if not key.lower().startswith("utm_") + and key.lower() not in _TRACKING_QUERY_PARAMS + ] + path = parsed.path or "/" + if path != "/": + path = path.rstrip("/") + return urlunsplit( + ( + parsed.scheme.lower(), + parsed.netloc.lower(), + path, + urlencode(sorted(query), doseq=True), + "", + ) + ) + + +def extract_exchange_ticker_hints(text: str) -> tuple[NewsTickerHint, ...]: + """Extract exchange-qualified ticker mentions from PR-style text. + + Examples matched include ``(NASDAQ: NVDA)``, ``NYSE: IBM``, and a + parenthesized comma list such as ``(NYSE: EVEX, EVEXW)``. A shared exchange + applies only to comma-separated symbols before the closing parenthesis; + semicolons and conjunctions end the group. Additional members use a + reconstructed ``EXCHANGE: SYMBOL`` raw value. The result is only a hint; + downstream instrument resolution should still validate it. + Markdown link and emphasis decoration is removed from a scan-only copy so + stored news text and its content hash retain their original representation. + """ + scan_text = _MARKDOWN_LINK_RE.sub(r"\1", text) + scan_text = _MARKDOWN_EMPHASIS_RE.sub(r"\1", scan_text) + hints: list[NewsTickerHint] = [] + seen: set[tuple[str, str | None]] = set() + for match in _EXCHANGE_TICKER_RE.finditer(scan_text): + raw_exchange = " ".join(match.group(1).upper().split()) + exchange = _EXCHANGE_NAMES.get(raw_exchange, raw_exchange) + symbol = match.group(2).upper() + key = (symbol, exchange) + if key in seen: + continue + seen.add(key) + hints.append( + NewsTickerHint( + symbol=symbol, + exchange=exchange, + raw=match.group(0).strip(), + ) + ) + if not match.group(0).lstrip().startswith("("): + continue + + closing_parenthesis = scan_text.find(")", match.end()) + if closing_parenthesis == -1: + continue + suffix = scan_text[match.end() : closing_parenthesis] + position = 0 + while continuation := _SHARED_EXCHANGE_SYMBOL_RE.match( + suffix, position + ): + symbol = continuation.group(1).upper() + key = (symbol, exchange) + if key not in seen: + seen.add(key) + hints.append( + NewsTickerHint( + symbol=symbol, + exchange=exchange, + raw=f"{exchange}: {symbol}", + ) + ) + position = continuation.end() + return tuple(hints) + + +async def _text_from_fetched(raw: Fetched) -> str: + ct = (raw.content_type or "").lower() + text = raw.bytes.decode("utf-8", errors="replace") + if ct.startswith("text/html") or ct.startswith("application/xhtml+xml"): + return await html_to_markdown(text) + if ( + ct.startswith("text/plain") + or ct.startswith("text/markdown") + or ct.startswith("application/xml") + or ct.startswith("text/xml") + ): + return text + raise ValueError(f"Unsupported content-type for news input: {ct!r}") + + +async def _html_or_text_to_markdown(value: str) -> str: + text = value.strip() + if not text: + return "" + if "<" not in text or ">" not in text: + return text + markdown = await html_to_markdown(text) + return markdown or text diff --git a/tests/preprocess/test_news.py b/tests/preprocess/test_news.py index 7a0b634..a05c13b 100644 --- a/tests/preprocess/test_news.py +++ b/tests/preprocess/test_news.py @@ -1,351 +1,390 @@ -"""Tests for preprocess.news — PR/wire candidate normalisation.""" - -import hashlib -import unittest -from datetime import datetime, timedelta, timezone -from pathlib import Path - -import httpx -import respx - -from quantmind.preprocess.fetch.rss import FeedItem -from quantmind.preprocess.news import ( - RawNewsDocument, - build_news_identity, - build_sec_news_identity, - canonicalize_source_url, - extract_exchange_ticker_hints, - feed_item_to_news_document, - fetch_news_document, - news_content_hash, - normalize_news_text, - preprocess_feed_item, - preprocess_news_document, -) -from quantmind.preprocess.time import parse_news_datetime - -_FIXTURES = Path(__file__).parent / "fixtures" / "pr_newswire" - - -class NewsTimeTests(unittest.TestCase): - def test_parse_rfc822_news_datetime(self): - result = parse_news_datetime("Mon, 15 Apr 2024 10:30:00 GMT") - - self.assertEqual(result.tzinfo, timezone.utc) - self.assertEqual(result.year, 2024) - self.assertEqual(result.hour, 10) - - def test_parse_news_datetime_with_offset(self): - result = parse_news_datetime("Mon, 15 Apr 2024 10:30:00 -0400") - - self.assertEqual(result.tzinfo, timezone.utc) - self.assertEqual(result.hour, 14) - - -class NewsPreprocessTests(unittest.TestCase): - def test_normalize_news_text_and_hash(self): - text = ( - "NVIDIA\u2014reported revenue\n\n\nNVIDIA\u2014reported revenue" - ) - normalized = normalize_news_text(text) - - self.assertEqual( - normalized, - "NVIDIA-reported revenue\n\nNVIDIA-reported revenue", - ) - self.assertEqual( - news_content_hash(normalized), - hashlib.sha256(normalized.encode("utf-8")).hexdigest(), - ) - - def test_normalize_news_text_stabilizes_email_protection_links(self): - text_a = ( - "Contact [[email protected]](/cdn-cgi/l/email-protection#abc123)" - ) - text_b = ( - "Contact [[email protected]](/cdn-cgi/l/email-protection#def456)" - ) - - self.assertEqual( - normalize_news_text(text_a), normalize_news_text(text_b) - ) - - def test_canonicalize_source_url_drops_tracking(self): - result = canonicalize_source_url( - "HTTPS://Example.COM/path/?b=2&utm_source=x&a=1#frag" - ) - - self.assertEqual(result, "https://example.com/path?a=1&b=2") - - def test_exchange_ticker_hints_are_deduped(self): - hints = extract_exchange_ticker_hints( - "NVIDIA Corporation (NASDAQ: NVDA) and IBM NYSE: IBM. " - "Again (NASDAQ: NVDA)." - ) - - self.assertEqual([h.symbol for h in hints], ["NVDA", "IBM"]) - self.assertEqual(hints[0].exchange, "NASDAQ") - self.assertEqual(hints[1].exchange, "NYSE") - - def test_exchange_ticker_hints_ignore_markdown_decoration(self): - cases = ( - ( - "link", - "Carnival Corporation & plc " - "(NYSE: [CCL](#financial-modal)) today announced...", - ("CCL", "NYSE", "(NYSE: CCL)"), - ), - ( - "bold emphasis", - "Example Corporation (NASDAQ: **ABC**) today announced...", - ("ABC", "NASDAQ", "(NASDAQ: ABC)"), - ), - ( - "italic emphasis", - "Example Corporation (NASDAQ: *ABC*) today announced...", - ("ABC", "NASDAQ", "(NASDAQ: ABC)"), - ), - ) - - for name, text, expected in cases: - with self.subTest(name=name): - hints = extract_exchange_ticker_hints(text) - - self.assertEqual(len(hints), 1) - self.assertEqual( - (hints[0].symbol, hints[0].exchange, hints[0].raw), - expected, - ) - - def test_build_sec_news_identity(self): - self.assertEqual( - build_sec_news_identity( - accession_number="0001045810-26-000123", - section_key="EX99.1", - ), - "sec:0001045810-26-000123:ex99.1", - ) - - def test_build_news_identity_requires_source_reference(self): - with self.assertRaises(ValueError): - build_news_identity(source_type="press_release") - - def test_preprocess_news_document_builds_candidate_contract(self): - published = datetime( - 2024, 4, 15, 10, 30, tzinfo=timezone(timedelta(hours=-4)) - ) - raw = RawNewsDocument( - body_text=( - "NVIDIA Corporation (NASDAQ: NVDA) today reported " - "record quarterly revenue." - ), - source_url="https://example.com/pr/nvidia-results?utm_source=feed", - title="NVIDIA Announces Results", - publisher="PR Newswire", - published_at=published, - payload_id="gnw-123", - ) - - candidate = preprocess_news_document(raw) - - self.assertEqual(candidate.source_type, "press_release") - self.assertTrue(candidate.identity.startswith("wire:")) - self.assertEqual( - candidate.source_url, "https://example.com/pr/nvidia-results" - ) - self.assertEqual(candidate.published_at.tzinfo, timezone.utc) - self.assertEqual(candidate.published_at.hour, 14) - self.assertEqual(candidate.ticker_hints[0].symbol, "NVDA") - self.assertEqual( - candidate.content_hash, - hashlib.sha256(candidate.body_text.encode("utf-8")).hexdigest(), - ) - - def test_preprocess_news_document_rejects_empty_body(self): - raw = RawNewsDocument( - body_text=" ", - source_url="https://example.com/pr/empty", - ) - - with self.assertRaises(ValueError): - preprocess_news_document(raw) - - def test_wire_markup_fixture_meets_ticker_recall_control(self): - body_text = (_FIXTURES / "ticker_markup.md").read_text(encoding="utf-8") - raw = RawNewsDocument( - body_text=body_text, - source_url="https://www.prnewswire.com/news-releases/example.html", - ) - - candidate = preprocess_news_document(raw) - - expected_hints = { - ("ABC", "NASDAQ"), - ("CCL", "NYSE"), - ("PODD", "NASDAQ"), - } - actual_hints = { - (hint.symbol, hint.exchange) for hint in candidate.ticker_hints - } - recall = len(expected_hints & actual_hints) / len(expected_hints) - self.assertEqual(recall, 1.0) - self.assertIn("[CCL](#financial-modal)", candidate.body_text) - self.assertIn( - "(NASDAQ:\n\n[PODD](#financial-modal))", - candidate.body_text, - ) - self.assertIn("**ABC**", candidate.body_text) - self.assertEqual( - candidate.content_hash, - news_content_hash(candidate.body_text), - ) - - -class NewsFeedItemTests(unittest.IsolatedAsyncioTestCase): - async def test_feed_item_to_news_document_uses_inline_content(self): - item = FeedItem( - title="NVIDIA Announces Results", - url="https://example.com/pr/nvidia-results", - id="gnw-123", - published_at=datetime(2024, 4, 15, tzinfo=timezone.utc), - content_html=( - "

NVIDIA Corporation " - "(NASDAQ: NVDA) reported results.

" - ), - source_feed_url="https://example.com/rss", - ) - - raw = await feed_item_to_news_document(item, publisher="PR Newswire") - - self.assertEqual(raw.payload_id, "gnw-123") - self.assertEqual(raw.publisher, "PR Newswire") - self.assertIn("NVIDIA", raw.body_text) - self.assertEqual( - raw.metadata["source_feed_url"], "https://example.com/rss" - ) - self.assertEqual(raw.metadata["body_source"], "feed") - - async def test_article_source_fetches_despite_nonempty_teaser(self): - item = FeedItem( - title="NVIDIA Announces Results", - url="https://example.com/pr/nvidia-results", - id="wire-123", - summary_html="

This non-empty teaser is not the full body.

", - source_feed_url="https://example.com/rss", - ) - html = """ -
-

NVIDIA Announces Results

-

The complete release contains full financial details.

-
- """ - with respx.mock(assert_all_called=True) as router: - router.get(item.url).mock( - return_value=httpx.Response( - 200, - headers={"Content-Type": "text/html"}, - text=html, - ) - ) - raw = await feed_item_to_news_document( - item, - body_source="article", - ) - - self.assertIn("complete release", raw.body_text) - self.assertNotIn("non-empty teaser", raw.body_text) - self.assertEqual(raw.metadata["body_source"], "article") - - async def test_feed_source_does_not_fetch_item_url(self): - item = FeedItem( - title="NVIDIA Announces Results", - url="https://example.com/pr/nvidia-results", - summary_html="

Feed body stays local.

", - ) - - with respx.mock(assert_all_called=True): - raw = await feed_item_to_news_document( - item, - body_source="feed", - ) - - self.assertIn("Feed body", raw.body_text) - - async def test_article_source_requires_item_url(self): - item = FeedItem(title="Missing URL", summary_html="A teaser") - - with self.assertRaisesRegex(ValueError, "requires an item URL"): - await feed_item_to_news_document( - item, - body_source="article", - ) - - async def test_article_extraction_failure_is_explicit(self): - item = FeedItem( - title="Unsupported body", - url="https://example.com/release.pdf", - summary_html="A teaser", - ) - with respx.mock(assert_all_called=True) as router: - router.get(item.url).mock( - return_value=httpx.Response( - 200, - headers={"Content-Type": "application/pdf"}, - content=b"not html", - ) - ) - with self.assertRaisesRegex(ValueError, "Unsupported"): - await feed_item_to_news_document( - item, - body_source="article", - ) - - async def test_preprocess_feed_item_returns_candidate(self): - item = FeedItem( - title="NVIDIA Announces Results", - url="https://example.com/pr/nvidia-results", - id="gnw-123", - summary_html="

NVIDIA Corporation (NASDAQ: NVDA) reported results.

", - ) - - candidate = await preprocess_feed_item(item, publisher="PR Newswire") - - self.assertEqual(candidate.publisher, "PR Newswire") - self.assertEqual(candidate.ticker_hints[0].symbol, "NVDA") - - -class FetchNewsDocumentTests(unittest.IsolatedAsyncioTestCase): - async def test_fetch_news_document_extracts_html(self): - html = """ - - -
-

NVIDIA Announces Results

-

NVIDIA Corporation (NASDAQ: NVDA) reported record revenue.

-
- - - """ - with respx.mock(assert_all_called=True) as router: - router.get("https://example.com/pr/nvidia-results").mock( - return_value=httpx.Response( - 200, - headers={"Content-Type": "text/html; charset=utf-8"}, - content=html.encode(), - ) - ) - raw = await fetch_news_document( - "https://example.com/pr/nvidia-results", - title="NVIDIA Announces Results", - payload_id="gnw-123", - ) - - self.assertEqual( - raw.source_url, "https://example.com/pr/nvidia-results" - ) - self.assertEqual(raw.metadata["content_type"], "text/html") - self.assertIn("record revenue", raw.body_text) - - -if __name__ == "__main__": - unittest.main() +"""Tests for preprocess.news — PR/wire candidate normalisation.""" + +import hashlib +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import httpx +import respx + +from quantmind.preprocess.fetch.rss import FeedItem +from quantmind.preprocess.news import ( + RawNewsDocument, + build_news_identity, + build_sec_news_identity, + canonicalize_source_url, + extract_exchange_ticker_hints, + feed_item_to_news_document, + fetch_news_document, + news_content_hash, + normalize_news_text, + preprocess_feed_item, + preprocess_news_document, +) +from quantmind.preprocess.time import parse_news_datetime + +_FIXTURES = Path(__file__).parent / "fixtures" / "pr_newswire" + + +class NewsTimeTests(unittest.TestCase): + def test_parse_rfc822_news_datetime(self): + result = parse_news_datetime("Mon, 15 Apr 2024 10:30:00 GMT") + + self.assertEqual(result.tzinfo, timezone.utc) + self.assertEqual(result.year, 2024) + self.assertEqual(result.hour, 10) + + def test_parse_news_datetime_with_offset(self): + result = parse_news_datetime("Mon, 15 Apr 2024 10:30:00 -0400") + + self.assertEqual(result.tzinfo, timezone.utc) + self.assertEqual(result.hour, 14) + + +class NewsPreprocessTests(unittest.TestCase): + def test_normalize_news_text_and_hash(self): + text = ( + "NVIDIA\u2014reported revenue\n\n\nNVIDIA\u2014reported revenue" + ) + normalized = normalize_news_text(text) + + self.assertEqual( + normalized, + "NVIDIA-reported revenue\n\nNVIDIA-reported revenue", + ) + self.assertEqual( + news_content_hash(normalized), + hashlib.sha256(normalized.encode("utf-8")).hexdigest(), + ) + + def test_normalize_news_text_stabilizes_email_protection_links(self): + text_a = ( + "Contact [[email protected]](/cdn-cgi/l/email-protection#abc123)" + ) + text_b = ( + "Contact [[email protected]](/cdn-cgi/l/email-protection#def456)" + ) + + self.assertEqual( + normalize_news_text(text_a), normalize_news_text(text_b) + ) + + def test_canonicalize_source_url_drops_tracking(self): + result = canonicalize_source_url( + "HTTPS://Example.COM/path/?b=2&utm_source=x&a=1#frag" + ) + + self.assertEqual(result, "https://example.com/path?a=1&b=2") + + def test_exchange_ticker_hints_are_deduped(self): + hints = extract_exchange_ticker_hints( + "NVIDIA Corporation (NASDAQ: NVDA) and IBM NYSE: IBM. " + "Again (NASDAQ: NVDA)." + ) + + self.assertEqual([h.symbol for h in hints], ["NVDA", "IBM"]) + self.assertEqual(hints[0].exchange, "NASDAQ") + self.assertEqual(hints[1].exchange, "NYSE") + + def test_exchange_ticker_hints_ignore_markdown_decoration(self): + cases = ( + ( + "link", + "Carnival Corporation & plc " + "(NYSE: [CCL](#financial-modal)) today announced...", + ("CCL", "NYSE", "(NYSE: CCL)"), + ), + ( + "bold emphasis", + "Example Corporation (NASDAQ: **ABC**) today announced...", + ("ABC", "NASDAQ", "(NASDAQ: ABC)"), + ), + ( + "italic emphasis", + "Example Corporation (NASDAQ: *ABC*) today announced...", + ("ABC", "NASDAQ", "(NASDAQ: ABC)"), + ), + ) + + for name, text, expected in cases: + with self.subTest(name=name): + hints = extract_exchange_ticker_hints(text) + + self.assertEqual(len(hints), 1) + self.assertEqual( + (hints[0].symbol, hints[0].exchange, hints[0].raw), + expected, + ) + + def test_exchange_ticker_hints_capture_shared_prefix_comma_lists(self): + cases = ( + ( + "supported shared prefix", + "(NYSE: EVEX, EVEXW; B3: EVEB31)", + ( + ("EVEX", "NYSE", "(NYSE: EVEX"), + ("EVEXW", "NYSE", "NYSE: EVEXW"), + ), + ), + ("unsupported exchange", "(OTCID: QVCAQ, QVCGQ, QVCPQ)", ()), + ( + "conjunction boundary", + "(NYSE: TME and HKEX: 1698)", + (("TME", "NYSE", "(NYSE: TME"),), + ), + ( + "semicolon boundary TSXV", + "(NASDAQ: VMAR; TSXV: VMAR)", + (("VMAR", "NASDAQ", "(NASDAQ: VMAR"),), + ), + ( + "semicolon boundary BMV", + "(NYSE: ASR; BMV: ASUR)", + (("ASR", "NYSE", "(NYSE: ASR"),), + ), + ) + + for name, text, expected in cases: + with self.subTest(name=name): + hints = extract_exchange_ticker_hints(text) + + self.assertEqual( + tuple( + (hint.symbol, hint.exchange, hint.raw) for hint in hints + ), + expected, + ) + + def test_build_sec_news_identity(self): + self.assertEqual( + build_sec_news_identity( + accession_number="0001045810-26-000123", + section_key="EX99.1", + ), + "sec:0001045810-26-000123:ex99.1", + ) + + def test_build_news_identity_requires_source_reference(self): + with self.assertRaises(ValueError): + build_news_identity(source_type="press_release") + + def test_preprocess_news_document_builds_candidate_contract(self): + published = datetime( + 2024, 4, 15, 10, 30, tzinfo=timezone(timedelta(hours=-4)) + ) + raw = RawNewsDocument( + body_text=( + "NVIDIA Corporation (NASDAQ: NVDA) today reported " + "record quarterly revenue." + ), + source_url="https://example.com/pr/nvidia-results?utm_source=feed", + title="NVIDIA Announces Results", + publisher="PR Newswire", + published_at=published, + payload_id="gnw-123", + ) + + candidate = preprocess_news_document(raw) + + self.assertEqual(candidate.source_type, "press_release") + self.assertTrue(candidate.identity.startswith("wire:")) + self.assertEqual( + candidate.source_url, "https://example.com/pr/nvidia-results" + ) + self.assertEqual(candidate.published_at.tzinfo, timezone.utc) + self.assertEqual(candidate.published_at.hour, 14) + self.assertEqual(candidate.ticker_hints[0].symbol, "NVDA") + self.assertEqual( + candidate.content_hash, + hashlib.sha256(candidate.body_text.encode("utf-8")).hexdigest(), + ) + + def test_preprocess_news_document_rejects_empty_body(self): + raw = RawNewsDocument( + body_text=" ", + source_url="https://example.com/pr/empty", + ) + + with self.assertRaises(ValueError): + preprocess_news_document(raw) + + def test_wire_markup_fixture_meets_ticker_recall_control(self): + body_text = (_FIXTURES / "ticker_markup.md").read_text(encoding="utf-8") + raw = RawNewsDocument( + body_text=body_text, + source_url="https://www.prnewswire.com/news-releases/example.html", + ) + + candidate = preprocess_news_document(raw) + + expected_hints = { + ("ABC", "NASDAQ"), + ("CCL", "NYSE"), + ("PODD", "NASDAQ"), + } + actual_hints = { + (hint.symbol, hint.exchange) for hint in candidate.ticker_hints + } + recall = len(expected_hints & actual_hints) / len(expected_hints) + self.assertEqual(recall, 1.0) + self.assertIn("[CCL](#financial-modal)", candidate.body_text) + self.assertIn( + "(NASDAQ:\n\n[PODD](#financial-modal))", + candidate.body_text, + ) + self.assertIn("**ABC**", candidate.body_text) + self.assertEqual( + candidate.content_hash, + news_content_hash(candidate.body_text), + ) + + +class NewsFeedItemTests(unittest.IsolatedAsyncioTestCase): + async def test_feed_item_to_news_document_uses_inline_content(self): + item = FeedItem( + title="NVIDIA Announces Results", + url="https://example.com/pr/nvidia-results", + id="gnw-123", + published_at=datetime(2024, 4, 15, tzinfo=timezone.utc), + content_html=( + "

NVIDIA Corporation " + "(NASDAQ: NVDA) reported results.

" + ), + source_feed_url="https://example.com/rss", + ) + + raw = await feed_item_to_news_document(item, publisher="PR Newswire") + + self.assertEqual(raw.payload_id, "gnw-123") + self.assertEqual(raw.publisher, "PR Newswire") + self.assertIn("NVIDIA", raw.body_text) + self.assertEqual( + raw.metadata["source_feed_url"], "https://example.com/rss" + ) + self.assertEqual(raw.metadata["body_source"], "feed") + + async def test_article_source_fetches_despite_nonempty_teaser(self): + item = FeedItem( + title="NVIDIA Announces Results", + url="https://example.com/pr/nvidia-results", + id="wire-123", + summary_html="

This non-empty teaser is not the full body.

", + source_feed_url="https://example.com/rss", + ) + html = """ +
+

NVIDIA Announces Results

+

The complete release contains full financial details.

+
+ """ + with respx.mock(assert_all_called=True) as router: + router.get(item.url).mock( + return_value=httpx.Response( + 200, + headers={"Content-Type": "text/html"}, + text=html, + ) + ) + raw = await feed_item_to_news_document( + item, + body_source="article", + ) + + self.assertIn("complete release", raw.body_text) + self.assertNotIn("non-empty teaser", raw.body_text) + self.assertEqual(raw.metadata["body_source"], "article") + + async def test_feed_source_does_not_fetch_item_url(self): + item = FeedItem( + title="NVIDIA Announces Results", + url="https://example.com/pr/nvidia-results", + summary_html="

Feed body stays local.

", + ) + + with respx.mock(assert_all_called=True): + raw = await feed_item_to_news_document( + item, + body_source="feed", + ) + + self.assertIn("Feed body", raw.body_text) + + async def test_article_source_requires_item_url(self): + item = FeedItem(title="Missing URL", summary_html="A teaser") + + with self.assertRaisesRegex(ValueError, "requires an item URL"): + await feed_item_to_news_document( + item, + body_source="article", + ) + + async def test_article_extraction_failure_is_explicit(self): + item = FeedItem( + title="Unsupported body", + url="https://example.com/release.pdf", + summary_html="A teaser", + ) + with respx.mock(assert_all_called=True) as router: + router.get(item.url).mock( + return_value=httpx.Response( + 200, + headers={"Content-Type": "application/pdf"}, + content=b"not html", + ) + ) + with self.assertRaisesRegex(ValueError, "Unsupported"): + await feed_item_to_news_document( + item, + body_source="article", + ) + + async def test_preprocess_feed_item_returns_candidate(self): + item = FeedItem( + title="NVIDIA Announces Results", + url="https://example.com/pr/nvidia-results", + id="gnw-123", + summary_html="

NVIDIA Corporation (NASDAQ: NVDA) reported results.

", + ) + + candidate = await preprocess_feed_item(item, publisher="PR Newswire") + + self.assertEqual(candidate.publisher, "PR Newswire") + self.assertEqual(candidate.ticker_hints[0].symbol, "NVDA") + + +class FetchNewsDocumentTests(unittest.IsolatedAsyncioTestCase): + async def test_fetch_news_document_extracts_html(self): + html = """ + + +
+

NVIDIA Announces Results

+

NVIDIA Corporation (NASDAQ: NVDA) reported record revenue.

+
+ + + """ + with respx.mock(assert_all_called=True) as router: + router.get("https://example.com/pr/nvidia-results").mock( + return_value=httpx.Response( + 200, + headers={"Content-Type": "text/html; charset=utf-8"}, + content=html.encode(), + ) + ) + raw = await fetch_news_document( + "https://example.com/pr/nvidia-results", + title="NVIDIA Announces Results", + payload_id="gnw-123", + ) + + self.assertEqual( + raw.source_url, "https://example.com/pr/nvidia-results" + ) + self.assertEqual(raw.metadata["content_type"], "text/html") + self.assertIn("record revenue", raw.body_text) + + +if __name__ == "__main__": + unittest.main() From 94931e55795e069d59b55ab7e862b7c1032bb674 Mon Sep 17 00:00:00 2001 From: pkuwkl Date: Thu, 23 Jul 2026 23:01:52 +0800 Subject: [PATCH 2/2] chore: normalize CRLF line endings to LF The files on this branch were committed with CRLF line terminators, which marked every line as changed and produced spurious merge conflicts with master. Convert them back to LF so the diff reflects only the real behavior change from #108. Co-Authored-By: Claude Opus 4.8 (1M context) --- contexts/design/flow/news.md | 566 ++++++------ examples/preprocess/02_news_ticker_lists.py | 16 +- quantmind/preprocess/news.py | 968 ++++++++++---------- tests/preprocess/test_news.py | 780 ++++++++-------- 4 files changed, 1165 insertions(+), 1165 deletions(-) diff --git a/contexts/design/flow/news.md b/contexts/design/flow/news.md index 7c365e1..fe6b852 100644 --- a/contexts/design/flow/news.md +++ b/contexts/design/flow/news.md @@ -1,283 +1,283 @@ -# News Collection Design - -## Quick Summary - -- **Purpose**: Define how QuantMind collects public news and records proof of what each source returned. -- **Read when**: Changing `collect_news`, `NewsWindow`, a news source, the `complete` flag, or news checks. -- **Status**: Current design; the open-source package currently supports PR Newswire. -- **Core rule**: Collection returns documents, failures, and whether the full time window was covered. Other code turns documents into knowledge, stores them, schedules runs, and removes irrelevant news. - -## Contents - -- [Scope and Current Support](#scope-and-current-support) -- [Design Principles](#design-principles) -- [Public API](#public-api) -- [Returned Data](#returned-data) -- [Collection Steps](#collection-steps) -- [Failures and the `complete` Flag](#failures-and-the-complete-flag) -- [Who Owns What](#who-owns-what) -- [How to Verify](#how-to-verify) -- [Out of Scope and When to Add Shared Code](#out-of-scope-and-when-to-add-shared-code) -- [Adding a Public News Source](#adding-a-public-news-source) - -## Scope and Current Support - -This page defines how the open-source package collects public company news. The -first version supports PR Newswire only. It has one collection function, one -time-window input, repeatable HTML cleanup, and visible item failures. - -Its public name follows the -[operation naming rules](../operations/naming.md): collection returns the news -as published plus fetch details. A separate operation turns those documents -into structured knowledge. - -The primary requirement is that any caller can request a complete, one-shot -collection of a past time window. A daily poll is therefore not a separate -operation; it is simply a short window evaluated on a schedule. - -## Design Principles - -1. **Ask for news, not fetch details.** Callers choose a source and time window. - They do not choose RSS, listing pages, how to move between pages, or article - parsing rules. -2. **Return every source row.** QuantMind does not silently remove duplicate - source rows. Repeated rows remain repeated output records, - and may share the same stable ID. -3. **Show partial failures.** The returned batch includes item failures. Invalid - inputs still raise before network work starts. -4. **Hide source implementation details.** PR Newswire may later use a public - mechanism other than listing pages without changing the public function. -5. **List supported sources explicitly.** The first version selects from a - closed source list. It does not expose a provider plugin API or registry. -6. **Keep downstream choices separate.** The calling pipeline owns storage, - duplicate handling, relevance rules, output formats, and scheduling. - -## Public API - -Callers use one entry point: - -```python -from datetime import datetime, timezone - -from quantmind.configs import NewsCollectionCfg, NewsWindow -from quantmind.flows import collect_news - -batch = await collect_news( - NewsWindow( - source="pr-newswire", - start=datetime(2026, 7, 13, tzinfo=timezone.utc), - end=datetime(2026, 7, 14, tzinfo=timezone.utc), - ), - cfg=NewsCollectionCfg(retain_raw_html=False), -) -``` - -`NewsWindow` requires timezone-aware timestamps. Its `[start, end)` interval -includes `start` and excludes `end`. Regular collection and historical runs -use the same call; there are no separate `poll_*`, `backfill_*`, or -`fetch_wire_*` public functions. - -`NewsCollectionCfg` keeps the shared `BaseFlowCfg` fields so it works with the -repository's shared config handling. Its only -collection-specific field is `retain_raw_html`, which controls whether fetched -article HTML bytes remain in the result. It defaults to `False`, which is the -storage-efficient behavior. The article is still fetched, hashed, parsed, and -described by fetch details; only its byte payload is discarded. The collector -does not otherwise use the shared model, tracing, or SDK-run fields. - -### Collection records are not knowledge - -QuantMind keeps collected documents separate from structured knowledge: - -| Operation | Result | Owning package | -|-----------|--------|-----------------| -| `collect_news` | Source documents, fetch details, failures, and whether the full window was scanned | `quantmind.preprocess` | -| future `extract_news_knowledge` | Extracted financial events | `quantmind.knowledge.News` | - -`NewsDocument` is therefore not a `KnowledgeItem`. It carries HTTP details, -raw bytes, parsing output, and collection status that remain useful before any -LLM or business data format is chosen. `knowledge.News` is a structured -financial event with entities, sentiment, financial importance -(`materiality`), source links, and text for embeddings. A pipeline may use both -types, but one cannot replace the other. - -## Returned Data - -`collect_news` returns a `NewsBatch` with four fields. Import these public types -from `quantmind.preprocess`: - -- `documents`: successfully collected `NewsDocument` observations; -- `failures`: lightweight `NewsFailure` records for work that could not be - completed; -- `observed_count`: the number of source rows successfully converted into - in-window observations, before article processing; -- `complete`: whether the listing scan covered the full requested window. - -A `NewsDocument` contains the source name, stable ID, preferred article URL, -title, publisher, publication time, cleaned Markdown, content hash, ticker -hints, and two `NewsArtifact` records: - -- a small record of the public listing row; -- an article record containing fetch details and a content hash. Its - `bytes` field is `None` unless `retain_raw_html=True`. - -`NewsArtifact` records what was fetched. It contains the content hash, -content type, source and resolved URLs, status, headers, and fetch time, with -an optional byte payload. - -Repeated listing rows are not removed. If two rows point to the same release, -the batch contains two observations with the same stable ID. A consumer can -use that ID to update the same database row safely while still recording what -QuantMind observed. - -## Collection Steps - -```text -NewsWindow - -> select the source collector - -> scan public listing pages from newest to oldest - -> keep rows inside [start, end) - -> fetch each linked article - -> convert HTML to Markdown - -> NewsDocument or NewsFailure - -> NewsBatch -``` - -PR Newswire discovery is based on its public news-release listing rather than -the latest-items RSS snapshot. Pages are read newest to oldest until an -observation strictly older than the requested start is seen. The strict -boundary matters because several rows with the same minute-level timestamp may -span two pages. A caller can therefore rerun a past-day request without a -previously saved cursor. - -PR Newswire exposes listing timestamps at minute precision. Scheduled callers -should therefore use minute-aligned window bounds, as the live end-to-end check -does. - -RSS remains an internal parser and a live check. It is not a public -`NewsInput`: choosing a feed is an implementation detail, and a limited feed -snapshot cannot prove that it covered a complete time window. - -The HTTP layer limits retries, waits between attempts, honors `Retry-After`, -and limits requests per host. PR Newswire-specific URL construction and -listing HTML parsing stay in the PR Newswire source module. - -`NewsWindow.source` lists every supported source. `collect_news` has an -explicit branch for each one, so the type checker catches a source name added -without a matching collector. - -## Failures and the `complete` Flag - -Configuration errors raise immediately. Examples include a naive timestamp, -an empty source, an unsupported source, or `end <= start`. - -After collection begins, one item failure does not stop independent items. -Each `NewsFailure` records the source, failed step, URL, optional item ID, error -category, and message. - -`complete=False` when any of the following is true: - -- a listing page could not be fetched or parsed; -- the listing scan stopped before crossing the window start. - -Article failures stay in `failures` but do not change whether all listing rows -were found. A caller can distinguish "the full time window was scanned" from -"every found article was processed." It can store successful records and retry -failed articles separately. An empty batch is complete only when the listing -scan crossed the requested start. - -## Who Owns What - -QuantMind owns: - -- scanning public listings and fetching articles; -- repeatable HTML cleanup; -- stable IDs and content hashes; -- honest batch counts, failure records, and completeness. - -The consuming production pipeline owns: - -- its schedule and when its GitHub Action runs; -- database storage, saved progress, and safe repeated writes; -- rules for removing duplicates; -- rules that remove irrelevant company news; -- later data formats, added fields, and database writes; -- shared batch callbacks, metrics, and monitoring. - -This split lets a separate ingestion job request the last day in one call, -apply its own rules, and write its own data format without adding those choices -to the open-source library. - -## How to Verify - -Use separate offline and live checks. - -### Offline repository checks - -`bash scripts/verify.sh` runs repeatable unit tests, linting, typing, and -coverage. News tests use saved HTML/RSS files and mocked HTTP responses. -They cover window validation, time edges, multi-page listings, duplicate -observations, raw-byte retention, retries, partial failures, and completeness. - -### Live end-to-end news check - -`python scripts/verify_news_e2e.py` is the news live-network check. It performs -three limited checks: - -1. fetch and parse the official PR Newswire RSS feed; -2. scan PR Newswire listing rows for the preceding 24 hours and prove that the - scan crossed the window start; -3. fetch up to 25 unique articles and find the first supported exchange-coded - symbol in every group that has one (100% recall). - -Ticker-hint extraction supports one exchange-qualified symbol and parenthesized -comma-separated symbol lists that share a supported exchange prefix. A shared -group ends at a semicolon, conjunction, or closing parenthesis. Additional list -members record a reconstructed `EXCHANGE: SYMBOL` raw value; unsupported -exchanges remain outside the whitelist policy. - -The ticker check uses a separate comparison regex rather than the production -extractor. It accepts plain, Markdown-link, and emphasis-wrapped symbols. It -recognizes comma-separated symbols that share a supported exchange prefix and -ignores exchanges that the extractor does not support. The script prints -PASS/FAIL plus short listing, article, -failure, and recall summaries. It fails when RSS or listing data is invalid, -no sampled article parses, or supported symbols fall below 100% recall. A -parsed sample with no supported symbols reports `SKIP` and still passes. - -The `news` job in `.github/workflows/e2e.yml` runs this check once daily, when -started manually, and on pull requests that change its direct dependencies. -The required `.github/workflows/ci.yml` workflow remains network-free so local -development and unit tests remain repeatable. - -## Out of Scope and When to Add Shared Code - -The first version does not include GlobeNewswire, Business Wire, authenticated -feeds, continuous cursors, storage, duplicate removal, financial importance -scoring, or a generic batch-operation base class. - -When a second source is implemented, compare its real behavior with PR -Newswire first. Add a shared collector interface only for behavior that both -working collectors actually share. Keep the public -`collect_news(NewsWindow, *, cfg)` call unchanged. - -## Adding a Public News Source - -A coding agent adding a second source follows this checklist: - -1. Add the source name to `NewsWindow.source`. -2. Add one private `quantmind/preprocess/.py` collector. -3. Add one explicit branch to `collect_news`; the type checker must pass. -4. Add saved-input tests for success, time edges, duplicate observations, - partial failures, and the `complete` flag. -5. Add a test proving that only the selected collector is called. -6. Update the supported-source table in `docs/README.md`, this design, and the - focused example if its common path changes. -7. Add or extend a component-specific live verifier and its named job in the - existing `.github/workflows/e2e.yml` when the integration depends on a - public network endpoint. Add its command to `docs/README.md`; do not add the - command to root agent guidance. - -Only after two real collectors share behavior should they use a common Python -`Protocol`. Never add a source only to the input `Literal`; the type checker is -expected to reject that incomplete change. +# News Collection Design + +## Quick Summary + +- **Purpose**: Define how QuantMind collects public news and records proof of what each source returned. +- **Read when**: Changing `collect_news`, `NewsWindow`, a news source, the `complete` flag, or news checks. +- **Status**: Current design; the open-source package currently supports PR Newswire. +- **Core rule**: Collection returns documents, failures, and whether the full time window was covered. Other code turns documents into knowledge, stores them, schedules runs, and removes irrelevant news. + +## Contents + +- [Scope and Current Support](#scope-and-current-support) +- [Design Principles](#design-principles) +- [Public API](#public-api) +- [Returned Data](#returned-data) +- [Collection Steps](#collection-steps) +- [Failures and the `complete` Flag](#failures-and-the-complete-flag) +- [Who Owns What](#who-owns-what) +- [How to Verify](#how-to-verify) +- [Out of Scope and When to Add Shared Code](#out-of-scope-and-when-to-add-shared-code) +- [Adding a Public News Source](#adding-a-public-news-source) + +## Scope and Current Support + +This page defines how the open-source package collects public company news. The +first version supports PR Newswire only. It has one collection function, one +time-window input, repeatable HTML cleanup, and visible item failures. + +Its public name follows the +[operation naming rules](../operations/naming.md): collection returns the news +as published plus fetch details. A separate operation turns those documents +into structured knowledge. + +The primary requirement is that any caller can request a complete, one-shot +collection of a past time window. A daily poll is therefore not a separate +operation; it is simply a short window evaluated on a schedule. + +## Design Principles + +1. **Ask for news, not fetch details.** Callers choose a source and time window. + They do not choose RSS, listing pages, how to move between pages, or article + parsing rules. +2. **Return every source row.** QuantMind does not silently remove duplicate + source rows. Repeated rows remain repeated output records, + and may share the same stable ID. +3. **Show partial failures.** The returned batch includes item failures. Invalid + inputs still raise before network work starts. +4. **Hide source implementation details.** PR Newswire may later use a public + mechanism other than listing pages without changing the public function. +5. **List supported sources explicitly.** The first version selects from a + closed source list. It does not expose a provider plugin API or registry. +6. **Keep downstream choices separate.** The calling pipeline owns storage, + duplicate handling, relevance rules, output formats, and scheduling. + +## Public API + +Callers use one entry point: + +```python +from datetime import datetime, timezone + +from quantmind.configs import NewsCollectionCfg, NewsWindow +from quantmind.flows import collect_news + +batch = await collect_news( + NewsWindow( + source="pr-newswire", + start=datetime(2026, 7, 13, tzinfo=timezone.utc), + end=datetime(2026, 7, 14, tzinfo=timezone.utc), + ), + cfg=NewsCollectionCfg(retain_raw_html=False), +) +``` + +`NewsWindow` requires timezone-aware timestamps. Its `[start, end)` interval +includes `start` and excludes `end`. Regular collection and historical runs +use the same call; there are no separate `poll_*`, `backfill_*`, or +`fetch_wire_*` public functions. + +`NewsCollectionCfg` keeps the shared `BaseFlowCfg` fields so it works with the +repository's shared config handling. Its only +collection-specific field is `retain_raw_html`, which controls whether fetched +article HTML bytes remain in the result. It defaults to `False`, which is the +storage-efficient behavior. The article is still fetched, hashed, parsed, and +described by fetch details; only its byte payload is discarded. The collector +does not otherwise use the shared model, tracing, or SDK-run fields. + +### Collection records are not knowledge + +QuantMind keeps collected documents separate from structured knowledge: + +| Operation | Result | Owning package | +|-----------|--------|-----------------| +| `collect_news` | Source documents, fetch details, failures, and whether the full window was scanned | `quantmind.preprocess` | +| future `extract_news_knowledge` | Extracted financial events | `quantmind.knowledge.News` | + +`NewsDocument` is therefore not a `KnowledgeItem`. It carries HTTP details, +raw bytes, parsing output, and collection status that remain useful before any +LLM or business data format is chosen. `knowledge.News` is a structured +financial event with entities, sentiment, financial importance +(`materiality`), source links, and text for embeddings. A pipeline may use both +types, but one cannot replace the other. + +## Returned Data + +`collect_news` returns a `NewsBatch` with four fields. Import these public types +from `quantmind.preprocess`: + +- `documents`: successfully collected `NewsDocument` observations; +- `failures`: lightweight `NewsFailure` records for work that could not be + completed; +- `observed_count`: the number of source rows successfully converted into + in-window observations, before article processing; +- `complete`: whether the listing scan covered the full requested window. + +A `NewsDocument` contains the source name, stable ID, preferred article URL, +title, publisher, publication time, cleaned Markdown, content hash, ticker +hints, and two `NewsArtifact` records: + +- a small record of the public listing row; +- an article record containing fetch details and a content hash. Its + `bytes` field is `None` unless `retain_raw_html=True`. + +`NewsArtifact` records what was fetched. It contains the content hash, +content type, source and resolved URLs, status, headers, and fetch time, with +an optional byte payload. + +Repeated listing rows are not removed. If two rows point to the same release, +the batch contains two observations with the same stable ID. A consumer can +use that ID to update the same database row safely while still recording what +QuantMind observed. + +## Collection Steps + +```text +NewsWindow + -> select the source collector + -> scan public listing pages from newest to oldest + -> keep rows inside [start, end) + -> fetch each linked article + -> convert HTML to Markdown + -> NewsDocument or NewsFailure + -> NewsBatch +``` + +PR Newswire discovery is based on its public news-release listing rather than +the latest-items RSS snapshot. Pages are read newest to oldest until an +observation strictly older than the requested start is seen. The strict +boundary matters because several rows with the same minute-level timestamp may +span two pages. A caller can therefore rerun a past-day request without a +previously saved cursor. + +PR Newswire exposes listing timestamps at minute precision. Scheduled callers +should therefore use minute-aligned window bounds, as the live end-to-end check +does. + +RSS remains an internal parser and a live check. It is not a public +`NewsInput`: choosing a feed is an implementation detail, and a limited feed +snapshot cannot prove that it covered a complete time window. + +The HTTP layer limits retries, waits between attempts, honors `Retry-After`, +and limits requests per host. PR Newswire-specific URL construction and +listing HTML parsing stay in the PR Newswire source module. + +`NewsWindow.source` lists every supported source. `collect_news` has an +explicit branch for each one, so the type checker catches a source name added +without a matching collector. + +## Failures and the `complete` Flag + +Configuration errors raise immediately. Examples include a naive timestamp, +an empty source, an unsupported source, or `end <= start`. + +After collection begins, one item failure does not stop independent items. +Each `NewsFailure` records the source, failed step, URL, optional item ID, error +category, and message. + +`complete=False` when any of the following is true: + +- a listing page could not be fetched or parsed; +- the listing scan stopped before crossing the window start. + +Article failures stay in `failures` but do not change whether all listing rows +were found. A caller can distinguish "the full time window was scanned" from +"every found article was processed." It can store successful records and retry +failed articles separately. An empty batch is complete only when the listing +scan crossed the requested start. + +## Who Owns What + +QuantMind owns: + +- scanning public listings and fetching articles; +- repeatable HTML cleanup; +- stable IDs and content hashes; +- honest batch counts, failure records, and completeness. + +The consuming production pipeline owns: + +- its schedule and when its GitHub Action runs; +- database storage, saved progress, and safe repeated writes; +- rules for removing duplicates; +- rules that remove irrelevant company news; +- later data formats, added fields, and database writes; +- shared batch callbacks, metrics, and monitoring. + +This split lets a separate ingestion job request the last day in one call, +apply its own rules, and write its own data format without adding those choices +to the open-source library. + +## How to Verify + +Use separate offline and live checks. + +### Offline repository checks + +`bash scripts/verify.sh` runs repeatable unit tests, linting, typing, and +coverage. News tests use saved HTML/RSS files and mocked HTTP responses. +They cover window validation, time edges, multi-page listings, duplicate +observations, raw-byte retention, retries, partial failures, and completeness. + +### Live end-to-end news check + +`python scripts/verify_news_e2e.py` is the news live-network check. It performs +three limited checks: + +1. fetch and parse the official PR Newswire RSS feed; +2. scan PR Newswire listing rows for the preceding 24 hours and prove that the + scan crossed the window start; +3. fetch up to 25 unique articles and find the first supported exchange-coded + symbol in every group that has one (100% recall). + +Ticker-hint extraction supports one exchange-qualified symbol and parenthesized +comma-separated symbol lists that share a supported exchange prefix. A shared +group ends at a semicolon, conjunction, or closing parenthesis. Additional list +members record a reconstructed `EXCHANGE: SYMBOL` raw value; unsupported +exchanges remain outside the whitelist policy. + +The ticker check uses a separate comparison regex rather than the production +extractor. It accepts plain, Markdown-link, and emphasis-wrapped symbols. It +recognizes comma-separated symbols that share a supported exchange prefix and +ignores exchanges that the extractor does not support. The script prints +PASS/FAIL plus short listing, article, +failure, and recall summaries. It fails when RSS or listing data is invalid, +no sampled article parses, or supported symbols fall below 100% recall. A +parsed sample with no supported symbols reports `SKIP` and still passes. + +The `news` job in `.github/workflows/e2e.yml` runs this check once daily, when +started manually, and on pull requests that change its direct dependencies. +The required `.github/workflows/ci.yml` workflow remains network-free so local +development and unit tests remain repeatable. + +## Out of Scope and When to Add Shared Code + +The first version does not include GlobeNewswire, Business Wire, authenticated +feeds, continuous cursors, storage, duplicate removal, financial importance +scoring, or a generic batch-operation base class. + +When a second source is implemented, compare its real behavior with PR +Newswire first. Add a shared collector interface only for behavior that both +working collectors actually share. Keep the public +`collect_news(NewsWindow, *, cfg)` call unchanged. + +## Adding a Public News Source + +A coding agent adding a second source follows this checklist: + +1. Add the source name to `NewsWindow.source`. +2. Add one private `quantmind/preprocess/.py` collector. +3. Add one explicit branch to `collect_news`; the type checker must pass. +4. Add saved-input tests for success, time edges, duplicate observations, + partial failures, and the `complete` flag. +5. Add a test proving that only the selected collector is called. +6. Update the supported-source table in `docs/README.md`, this design, and the + focused example if its common path changes. +7. Add or extend a component-specific live verifier and its named job in the + existing `.github/workflows/e2e.yml` when the integration depends on a + public network endpoint. Add its command to `docs/README.md`; do not add the + command to root agent guidance. + +Only after two real collectors share behavior should they use a common Python +`Protocol`. Never add a source only to the input `Literal`; the type checker is +expected to reject that incomplete change. diff --git a/examples/preprocess/02_news_ticker_lists.py b/examples/preprocess/02_news_ticker_lists.py index 0700114..dfd7ddd 100644 --- a/examples/preprocess/02_news_ticker_lists.py +++ b/examples/preprocess/02_news_ticker_lists.py @@ -1,8 +1,8 @@ -"""Extract ticker hints from a shared exchange-prefix list.""" - -from quantmind.preprocess import extract_exchange_ticker_hints - -text = "Example issuer (NYSE: EVEX, EVEXW; B3: EVEB31) announced results." - -for hint in extract_exchange_ticker_hints(text): - print(hint) +"""Extract ticker hints from a shared exchange-prefix list.""" + +from quantmind.preprocess import extract_exchange_ticker_hints + +text = "Example issuer (NYSE: EVEX, EVEXW; B3: EVEB31) announced results." + +for hint in extract_exchange_ticker_hints(text): + print(hint) diff --git a/quantmind/preprocess/news.py b/quantmind/preprocess/news.py index fffcc65..314c999 100644 --- a/quantmind/preprocess/news.py +++ b/quantmind/preprocess/news.py @@ -1,484 +1,484 @@ -"""Source-agnostic news preprocessing and candidate normalization.""" - -import hashlib -import re -from dataclasses import dataclass, field -from datetime import datetime -from typing import Literal -from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit - -from quantmind.preprocess._news_types import NewsTickerHint -from quantmind.preprocess.clean import ( - collapse_whitespace, - dedupe_lines, - normalize_unicode, -) -from quantmind.preprocess.fetch._types import Fetched -from quantmind.preprocess.fetch.http import HttpFetcher, fetch_url -from quantmind.preprocess.fetch.rss import FeedItem -from quantmind.preprocess.format.html import html_to_markdown -from quantmind.preprocess.time import to_utc - -NewsSourceType = Literal[ - "company_8k", - "press_release", - "publisher_news", - "regulatory_news", -] -BodySource = Literal["feed", "article"] - -_SOURCE_PREFIX: dict[str, str] = { - "company_8k": "sec", - "press_release": "wire", - "publisher_news": "publisher", - "regulatory_news": "regulatory", -} - -_TRACKING_QUERY_PARAMS = { - "fbclid", - "feedref", - "gclid", - "mc_cid", - "mc_eid", - "spm", -} - -_EXCHANGE_TICKER_RE = re.compile( - r"(?:\(|\b)" - r"(NASDAQ|NYSE(?:\s+American|\s+MKT|\s+Arca)?|AMEX|OTCQX|OTCQB|OTC|CBOE)" - r"\s*:\s*" - r"([A-Z][A-Z0-9.-]{0,9})" - r"(?:\)|\b)", - re.IGNORECASE, -) -_MARKDOWN_LINK_RE = re.compile(r"\[([^\]\n]+)\]\([^\)\n]*\)") -_MARKDOWN_EMPHASIS_RE = re.compile( - r"(? RawNewsDocument: - """Fetch one news/press-release URL and extract plain markdown text. - - Args: - url: Article or press-release URL. - source_type: Normalised source family. - title: Optional title supplied by a feed or caller. - publisher: Optional source publisher. - published_at: Optional publication timestamp. - payload_id: Optional upstream feed/item id. - timeout: Per-request timeout in seconds. - max_bytes: Hard ceiling on response body size. - fetcher: Optional shared HTTP fetcher and host-rate state. - - Returns: - Raw document with extracted markdown text and caller metadata. - - Raises: - ValueError: If the content type cannot be converted to text. - httpx.HTTPError: For network / status / timeout failures. - """ - if fetcher is None: - fetched = await fetch_url(url, timeout=timeout, max_bytes=max_bytes) - else: - fetched = await fetcher.fetch_url( - url, - timeout=timeout, - max_bytes=max_bytes, - ) - return await news_document_from_fetched( - fetched, - source_type=source_type, - title=title, - publisher=publisher, - published_at=published_at, - payload_id=payload_id, - ) - - -async def news_document_from_fetched( - fetched: Fetched, - *, - source_type: NewsSourceType = "press_release", - title: str | None = None, - publisher: str | None = None, - published_at: datetime | None = None, - payload_id: str | None = None, -) -> RawNewsDocument: - """Extract a raw news document from already-fetched evidence.""" - body_text = await _text_from_fetched(fetched) - return RawNewsDocument( - body_text=body_text, - source_type=source_type, - source_url=fetched.resolved_url or fetched.source_url, - title=title, - publisher=publisher, - published_at=published_at, - payload_id=payload_id, - metadata={ - "body_source": "article", - "content_type": fetched.content_type, - }, - ) - - -async def preprocess_news_url( - url: str, - *, - source_type: NewsSourceType = "press_release", - title: str | None = None, - publisher: str | None = None, - published_at: datetime | None = None, - payload_id: str | None = None, - timeout: float = 30.0, - max_bytes: int = 10_000_000, -) -> NewsCandidate: - """Fetch and normalise one news/press-release URL.""" - raw = await fetch_news_document( - url, - source_type=source_type, - title=title, - publisher=publisher, - published_at=published_at, - payload_id=payload_id, - timeout=timeout, - max_bytes=max_bytes, - ) - return preprocess_news_document(raw) - - -async def feed_item_to_news_document( - item: FeedItem, - *, - source_type: NewsSourceType = "press_release", - publisher: str | None = None, - body_source: BodySource = "feed", - fetcher: HttpFetcher | None = None, -) -> RawNewsDocument: - """Convert one parsed RSS/Atom item into a raw news document. - - ``body_source="feed"`` trusts the RSS/Atom body. ``"article"`` always - follows the item URL, even when the feed contains a non-empty teaser. - """ - if body_source == "article": - if not item.url: - raise ValueError("article body_source requires an item URL") - raw = await fetch_news_document( - item.url, - source_type=source_type, - title=item.title, - publisher=publisher, - published_at=item.published_at, - payload_id=item.id, - fetcher=fetcher, - ) - return RawNewsDocument( - body_text=raw.body_text, - source_type=raw.source_type, - source_url=raw.source_url, - title=raw.title, - publisher=raw.publisher, - published_at=raw.published_at, - payload_id=raw.payload_id, - metadata={ - **raw.metadata, - **{ - key: value - for key, value in { - "source_feed_url": item.source_feed_url, - **item.raw, - }.items() - if value - }, - }, - ) - - body_text = await _html_or_text_to_markdown( - item.content_html or item.summary_html or "" - ) - return RawNewsDocument( - body_text=body_text, - source_type=source_type, - source_url=item.url, - title=item.title, - publisher=publisher, - published_at=item.published_at, - payload_id=item.id, - metadata={ - key: value - for key, value in { - "body_source": "feed", - "source_feed_url": item.source_feed_url, - **item.raw, - }.items() - if value - }, - ) - - -async def preprocess_feed_item( - item: FeedItem, - *, - source_type: NewsSourceType = "press_release", - publisher: str | None = None, - body_source: BodySource = "feed", - fetcher: HttpFetcher | None = None, -) -> NewsCandidate: - """Convert and normalise one parsed RSS/Atom item.""" - raw = await feed_item_to_news_document( - item, - source_type=source_type, - publisher=publisher, - body_source=body_source, - fetcher=fetcher, - ) - return preprocess_news_document(raw) - - -def preprocess_news_document(raw: RawNewsDocument) -> NewsCandidate: - """Normalise a raw news document into the shared candidate contract. - - The output carries stable source identity, cleaned body text, a content - hash, provenance, timestamp, and deterministic ticker hints. - """ - body_text = normalize_news_text(raw.body_text) - if not body_text: - raise ValueError("news document body is empty after preprocessing") - - source_url = ( - canonicalize_source_url(raw.source_url) if raw.source_url else None - ) - title = normalize_news_text(raw.title or "") or None - published_at = to_utc(raw.published_at) if raw.published_at else None - ticker_scan = "\n".join(part for part in (title, body_text) if part) - - return NewsCandidate( - body_text=body_text, - content_hash=news_content_hash(body_text), - source_type=raw.source_type, - identity=build_news_identity( - source_type=raw.source_type, - source_url=source_url, - payload_id=raw.payload_id, - ), - source_url=source_url, - title=title, - publisher=raw.publisher, - published_at=published_at, - ticker_hints=extract_exchange_ticker_hints(ticker_scan), - metadata=dict(raw.metadata), - ) - - -def normalize_news_text(text: str) -> str: - """Apply QuantMind's canonical text cleanup order for news bodies.""" - normalized = normalize_unicode(text) - normalized = _EMAIL_PROTECTION_LINK_RE.sub("[email protected]", normalized) - return collapse_whitespace(dedupe_lines(normalized)) - - -def news_content_hash(text: str) -> str: - """Return the sha256 hash for an already-normalised news body.""" - return hashlib.sha256(text.encode("utf-8")).hexdigest() - - -def build_news_identity( - *, - source_type: NewsSourceType, - source_url: str | None = None, - payload_id: str | None = None, -) -> str: - """Build a deterministic source-document identity. - - Press-release/wire rows use the payload id when available and otherwise - fall back to the canonical source URL. The identity is hashed so callers do - not accidentally depend on upstream id formatting. - """ - identity = (payload_id or "").strip() - if not identity and source_url: - identity = canonicalize_source_url(source_url) - if not identity: - raise ValueError("news identity requires payload_id or source_url") - prefix = _SOURCE_PREFIX[source_type] - digest = hashlib.sha256(identity.encode("utf-8")).hexdigest() - return f"{prefix}:{digest}" - - -def build_sec_news_identity( - *, - accession_number: str, - section_key: str, -) -> str: - """Build a stable 8-K EX-99.x news identity.""" - accession = accession_number.strip() - section = section_key.strip().lower() - if not accession or not section: - raise ValueError("SEC news identity requires accession and section") - return f"sec:{accession}:{section}" - - -def canonicalize_source_url(url: str) -> str: - """Normalise source URLs for stable source identity.""" - parsed = urlsplit(url.strip()) - query = [ - (key, value) - for key, value in parse_qsl(parsed.query, keep_blank_values=True) - if not key.lower().startswith("utm_") - and key.lower() not in _TRACKING_QUERY_PARAMS - ] - path = parsed.path or "/" - if path != "/": - path = path.rstrip("/") - return urlunsplit( - ( - parsed.scheme.lower(), - parsed.netloc.lower(), - path, - urlencode(sorted(query), doseq=True), - "", - ) - ) - - -def extract_exchange_ticker_hints(text: str) -> tuple[NewsTickerHint, ...]: - """Extract exchange-qualified ticker mentions from PR-style text. - - Examples matched include ``(NASDAQ: NVDA)``, ``NYSE: IBM``, and a - parenthesized comma list such as ``(NYSE: EVEX, EVEXW)``. A shared exchange - applies only to comma-separated symbols before the closing parenthesis; - semicolons and conjunctions end the group. Additional members use a - reconstructed ``EXCHANGE: SYMBOL`` raw value. The result is only a hint; - downstream instrument resolution should still validate it. - Markdown link and emphasis decoration is removed from a scan-only copy so - stored news text and its content hash retain their original representation. - """ - scan_text = _MARKDOWN_LINK_RE.sub(r"\1", text) - scan_text = _MARKDOWN_EMPHASIS_RE.sub(r"\1", scan_text) - hints: list[NewsTickerHint] = [] - seen: set[tuple[str, str | None]] = set() - for match in _EXCHANGE_TICKER_RE.finditer(scan_text): - raw_exchange = " ".join(match.group(1).upper().split()) - exchange = _EXCHANGE_NAMES.get(raw_exchange, raw_exchange) - symbol = match.group(2).upper() - key = (symbol, exchange) - if key in seen: - continue - seen.add(key) - hints.append( - NewsTickerHint( - symbol=symbol, - exchange=exchange, - raw=match.group(0).strip(), - ) - ) - if not match.group(0).lstrip().startswith("("): - continue - - closing_parenthesis = scan_text.find(")", match.end()) - if closing_parenthesis == -1: - continue - suffix = scan_text[match.end() : closing_parenthesis] - position = 0 - while continuation := _SHARED_EXCHANGE_SYMBOL_RE.match( - suffix, position - ): - symbol = continuation.group(1).upper() - key = (symbol, exchange) - if key not in seen: - seen.add(key) - hints.append( - NewsTickerHint( - symbol=symbol, - exchange=exchange, - raw=f"{exchange}: {symbol}", - ) - ) - position = continuation.end() - return tuple(hints) - - -async def _text_from_fetched(raw: Fetched) -> str: - ct = (raw.content_type or "").lower() - text = raw.bytes.decode("utf-8", errors="replace") - if ct.startswith("text/html") or ct.startswith("application/xhtml+xml"): - return await html_to_markdown(text) - if ( - ct.startswith("text/plain") - or ct.startswith("text/markdown") - or ct.startswith("application/xml") - or ct.startswith("text/xml") - ): - return text - raise ValueError(f"Unsupported content-type for news input: {ct!r}") - - -async def _html_or_text_to_markdown(value: str) -> str: - text = value.strip() - if not text: - return "" - if "<" not in text or ">" not in text: - return text - markdown = await html_to_markdown(text) - return markdown or text +"""Source-agnostic news preprocessing and candidate normalization.""" + +import hashlib +import re +from dataclasses import dataclass, field +from datetime import datetime +from typing import Literal +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + +from quantmind.preprocess._news_types import NewsTickerHint +from quantmind.preprocess.clean import ( + collapse_whitespace, + dedupe_lines, + normalize_unicode, +) +from quantmind.preprocess.fetch._types import Fetched +from quantmind.preprocess.fetch.http import HttpFetcher, fetch_url +from quantmind.preprocess.fetch.rss import FeedItem +from quantmind.preprocess.format.html import html_to_markdown +from quantmind.preprocess.time import to_utc + +NewsSourceType = Literal[ + "company_8k", + "press_release", + "publisher_news", + "regulatory_news", +] +BodySource = Literal["feed", "article"] + +_SOURCE_PREFIX: dict[str, str] = { + "company_8k": "sec", + "press_release": "wire", + "publisher_news": "publisher", + "regulatory_news": "regulatory", +} + +_TRACKING_QUERY_PARAMS = { + "fbclid", + "feedref", + "gclid", + "mc_cid", + "mc_eid", + "spm", +} + +_EXCHANGE_TICKER_RE = re.compile( + r"(?:\(|\b)" + r"(NASDAQ|NYSE(?:\s+American|\s+MKT|\s+Arca)?|AMEX|OTCQX|OTCQB|OTC|CBOE)" + r"\s*:\s*" + r"([A-Z][A-Z0-9.-]{0,9})" + r"(?:\)|\b)", + re.IGNORECASE, +) +_MARKDOWN_LINK_RE = re.compile(r"\[([^\]\n]+)\]\([^\)\n]*\)") +_MARKDOWN_EMPHASIS_RE = re.compile( + r"(? RawNewsDocument: + """Fetch one news/press-release URL and extract plain markdown text. + + Args: + url: Article or press-release URL. + source_type: Normalised source family. + title: Optional title supplied by a feed or caller. + publisher: Optional source publisher. + published_at: Optional publication timestamp. + payload_id: Optional upstream feed/item id. + timeout: Per-request timeout in seconds. + max_bytes: Hard ceiling on response body size. + fetcher: Optional shared HTTP fetcher and host-rate state. + + Returns: + Raw document with extracted markdown text and caller metadata. + + Raises: + ValueError: If the content type cannot be converted to text. + httpx.HTTPError: For network / status / timeout failures. + """ + if fetcher is None: + fetched = await fetch_url(url, timeout=timeout, max_bytes=max_bytes) + else: + fetched = await fetcher.fetch_url( + url, + timeout=timeout, + max_bytes=max_bytes, + ) + return await news_document_from_fetched( + fetched, + source_type=source_type, + title=title, + publisher=publisher, + published_at=published_at, + payload_id=payload_id, + ) + + +async def news_document_from_fetched( + fetched: Fetched, + *, + source_type: NewsSourceType = "press_release", + title: str | None = None, + publisher: str | None = None, + published_at: datetime | None = None, + payload_id: str | None = None, +) -> RawNewsDocument: + """Extract a raw news document from already-fetched evidence.""" + body_text = await _text_from_fetched(fetched) + return RawNewsDocument( + body_text=body_text, + source_type=source_type, + source_url=fetched.resolved_url or fetched.source_url, + title=title, + publisher=publisher, + published_at=published_at, + payload_id=payload_id, + metadata={ + "body_source": "article", + "content_type": fetched.content_type, + }, + ) + + +async def preprocess_news_url( + url: str, + *, + source_type: NewsSourceType = "press_release", + title: str | None = None, + publisher: str | None = None, + published_at: datetime | None = None, + payload_id: str | None = None, + timeout: float = 30.0, + max_bytes: int = 10_000_000, +) -> NewsCandidate: + """Fetch and normalise one news/press-release URL.""" + raw = await fetch_news_document( + url, + source_type=source_type, + title=title, + publisher=publisher, + published_at=published_at, + payload_id=payload_id, + timeout=timeout, + max_bytes=max_bytes, + ) + return preprocess_news_document(raw) + + +async def feed_item_to_news_document( + item: FeedItem, + *, + source_type: NewsSourceType = "press_release", + publisher: str | None = None, + body_source: BodySource = "feed", + fetcher: HttpFetcher | None = None, +) -> RawNewsDocument: + """Convert one parsed RSS/Atom item into a raw news document. + + ``body_source="feed"`` trusts the RSS/Atom body. ``"article"`` always + follows the item URL, even when the feed contains a non-empty teaser. + """ + if body_source == "article": + if not item.url: + raise ValueError("article body_source requires an item URL") + raw = await fetch_news_document( + item.url, + source_type=source_type, + title=item.title, + publisher=publisher, + published_at=item.published_at, + payload_id=item.id, + fetcher=fetcher, + ) + return RawNewsDocument( + body_text=raw.body_text, + source_type=raw.source_type, + source_url=raw.source_url, + title=raw.title, + publisher=raw.publisher, + published_at=raw.published_at, + payload_id=raw.payload_id, + metadata={ + **raw.metadata, + **{ + key: value + for key, value in { + "source_feed_url": item.source_feed_url, + **item.raw, + }.items() + if value + }, + }, + ) + + body_text = await _html_or_text_to_markdown( + item.content_html or item.summary_html or "" + ) + return RawNewsDocument( + body_text=body_text, + source_type=source_type, + source_url=item.url, + title=item.title, + publisher=publisher, + published_at=item.published_at, + payload_id=item.id, + metadata={ + key: value + for key, value in { + "body_source": "feed", + "source_feed_url": item.source_feed_url, + **item.raw, + }.items() + if value + }, + ) + + +async def preprocess_feed_item( + item: FeedItem, + *, + source_type: NewsSourceType = "press_release", + publisher: str | None = None, + body_source: BodySource = "feed", + fetcher: HttpFetcher | None = None, +) -> NewsCandidate: + """Convert and normalise one parsed RSS/Atom item.""" + raw = await feed_item_to_news_document( + item, + source_type=source_type, + publisher=publisher, + body_source=body_source, + fetcher=fetcher, + ) + return preprocess_news_document(raw) + + +def preprocess_news_document(raw: RawNewsDocument) -> NewsCandidate: + """Normalise a raw news document into the shared candidate contract. + + The output carries stable source identity, cleaned body text, a content + hash, provenance, timestamp, and deterministic ticker hints. + """ + body_text = normalize_news_text(raw.body_text) + if not body_text: + raise ValueError("news document body is empty after preprocessing") + + source_url = ( + canonicalize_source_url(raw.source_url) if raw.source_url else None + ) + title = normalize_news_text(raw.title or "") or None + published_at = to_utc(raw.published_at) if raw.published_at else None + ticker_scan = "\n".join(part for part in (title, body_text) if part) + + return NewsCandidate( + body_text=body_text, + content_hash=news_content_hash(body_text), + source_type=raw.source_type, + identity=build_news_identity( + source_type=raw.source_type, + source_url=source_url, + payload_id=raw.payload_id, + ), + source_url=source_url, + title=title, + publisher=raw.publisher, + published_at=published_at, + ticker_hints=extract_exchange_ticker_hints(ticker_scan), + metadata=dict(raw.metadata), + ) + + +def normalize_news_text(text: str) -> str: + """Apply QuantMind's canonical text cleanup order for news bodies.""" + normalized = normalize_unicode(text) + normalized = _EMAIL_PROTECTION_LINK_RE.sub("[email protected]", normalized) + return collapse_whitespace(dedupe_lines(normalized)) + + +def news_content_hash(text: str) -> str: + """Return the sha256 hash for an already-normalised news body.""" + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def build_news_identity( + *, + source_type: NewsSourceType, + source_url: str | None = None, + payload_id: str | None = None, +) -> str: + """Build a deterministic source-document identity. + + Press-release/wire rows use the payload id when available and otherwise + fall back to the canonical source URL. The identity is hashed so callers do + not accidentally depend on upstream id formatting. + """ + identity = (payload_id or "").strip() + if not identity and source_url: + identity = canonicalize_source_url(source_url) + if not identity: + raise ValueError("news identity requires payload_id or source_url") + prefix = _SOURCE_PREFIX[source_type] + digest = hashlib.sha256(identity.encode("utf-8")).hexdigest() + return f"{prefix}:{digest}" + + +def build_sec_news_identity( + *, + accession_number: str, + section_key: str, +) -> str: + """Build a stable 8-K EX-99.x news identity.""" + accession = accession_number.strip() + section = section_key.strip().lower() + if not accession or not section: + raise ValueError("SEC news identity requires accession and section") + return f"sec:{accession}:{section}" + + +def canonicalize_source_url(url: str) -> str: + """Normalise source URLs for stable source identity.""" + parsed = urlsplit(url.strip()) + query = [ + (key, value) + for key, value in parse_qsl(parsed.query, keep_blank_values=True) + if not key.lower().startswith("utm_") + and key.lower() not in _TRACKING_QUERY_PARAMS + ] + path = parsed.path or "/" + if path != "/": + path = path.rstrip("/") + return urlunsplit( + ( + parsed.scheme.lower(), + parsed.netloc.lower(), + path, + urlencode(sorted(query), doseq=True), + "", + ) + ) + + +def extract_exchange_ticker_hints(text: str) -> tuple[NewsTickerHint, ...]: + """Extract exchange-qualified ticker mentions from PR-style text. + + Examples matched include ``(NASDAQ: NVDA)``, ``NYSE: IBM``, and a + parenthesized comma list such as ``(NYSE: EVEX, EVEXW)``. A shared exchange + applies only to comma-separated symbols before the closing parenthesis; + semicolons and conjunctions end the group. Additional members use a + reconstructed ``EXCHANGE: SYMBOL`` raw value. The result is only a hint; + downstream instrument resolution should still validate it. + Markdown link and emphasis decoration is removed from a scan-only copy so + stored news text and its content hash retain their original representation. + """ + scan_text = _MARKDOWN_LINK_RE.sub(r"\1", text) + scan_text = _MARKDOWN_EMPHASIS_RE.sub(r"\1", scan_text) + hints: list[NewsTickerHint] = [] + seen: set[tuple[str, str | None]] = set() + for match in _EXCHANGE_TICKER_RE.finditer(scan_text): + raw_exchange = " ".join(match.group(1).upper().split()) + exchange = _EXCHANGE_NAMES.get(raw_exchange, raw_exchange) + symbol = match.group(2).upper() + key = (symbol, exchange) + if key in seen: + continue + seen.add(key) + hints.append( + NewsTickerHint( + symbol=symbol, + exchange=exchange, + raw=match.group(0).strip(), + ) + ) + if not match.group(0).lstrip().startswith("("): + continue + + closing_parenthesis = scan_text.find(")", match.end()) + if closing_parenthesis == -1: + continue + suffix = scan_text[match.end() : closing_parenthesis] + position = 0 + while continuation := _SHARED_EXCHANGE_SYMBOL_RE.match( + suffix, position + ): + symbol = continuation.group(1).upper() + key = (symbol, exchange) + if key not in seen: + seen.add(key) + hints.append( + NewsTickerHint( + symbol=symbol, + exchange=exchange, + raw=f"{exchange}: {symbol}", + ) + ) + position = continuation.end() + return tuple(hints) + + +async def _text_from_fetched(raw: Fetched) -> str: + ct = (raw.content_type or "").lower() + text = raw.bytes.decode("utf-8", errors="replace") + if ct.startswith("text/html") or ct.startswith("application/xhtml+xml"): + return await html_to_markdown(text) + if ( + ct.startswith("text/plain") + or ct.startswith("text/markdown") + or ct.startswith("application/xml") + or ct.startswith("text/xml") + ): + return text + raise ValueError(f"Unsupported content-type for news input: {ct!r}") + + +async def _html_or_text_to_markdown(value: str) -> str: + text = value.strip() + if not text: + return "" + if "<" not in text or ">" not in text: + return text + markdown = await html_to_markdown(text) + return markdown or text diff --git a/tests/preprocess/test_news.py b/tests/preprocess/test_news.py index a05c13b..4f380ee 100644 --- a/tests/preprocess/test_news.py +++ b/tests/preprocess/test_news.py @@ -1,390 +1,390 @@ -"""Tests for preprocess.news — PR/wire candidate normalisation.""" - -import hashlib -import unittest -from datetime import datetime, timedelta, timezone -from pathlib import Path - -import httpx -import respx - -from quantmind.preprocess.fetch.rss import FeedItem -from quantmind.preprocess.news import ( - RawNewsDocument, - build_news_identity, - build_sec_news_identity, - canonicalize_source_url, - extract_exchange_ticker_hints, - feed_item_to_news_document, - fetch_news_document, - news_content_hash, - normalize_news_text, - preprocess_feed_item, - preprocess_news_document, -) -from quantmind.preprocess.time import parse_news_datetime - -_FIXTURES = Path(__file__).parent / "fixtures" / "pr_newswire" - - -class NewsTimeTests(unittest.TestCase): - def test_parse_rfc822_news_datetime(self): - result = parse_news_datetime("Mon, 15 Apr 2024 10:30:00 GMT") - - self.assertEqual(result.tzinfo, timezone.utc) - self.assertEqual(result.year, 2024) - self.assertEqual(result.hour, 10) - - def test_parse_news_datetime_with_offset(self): - result = parse_news_datetime("Mon, 15 Apr 2024 10:30:00 -0400") - - self.assertEqual(result.tzinfo, timezone.utc) - self.assertEqual(result.hour, 14) - - -class NewsPreprocessTests(unittest.TestCase): - def test_normalize_news_text_and_hash(self): - text = ( - "NVIDIA\u2014reported revenue\n\n\nNVIDIA\u2014reported revenue" - ) - normalized = normalize_news_text(text) - - self.assertEqual( - normalized, - "NVIDIA-reported revenue\n\nNVIDIA-reported revenue", - ) - self.assertEqual( - news_content_hash(normalized), - hashlib.sha256(normalized.encode("utf-8")).hexdigest(), - ) - - def test_normalize_news_text_stabilizes_email_protection_links(self): - text_a = ( - "Contact [[email protected]](/cdn-cgi/l/email-protection#abc123)" - ) - text_b = ( - "Contact [[email protected]](/cdn-cgi/l/email-protection#def456)" - ) - - self.assertEqual( - normalize_news_text(text_a), normalize_news_text(text_b) - ) - - def test_canonicalize_source_url_drops_tracking(self): - result = canonicalize_source_url( - "HTTPS://Example.COM/path/?b=2&utm_source=x&a=1#frag" - ) - - self.assertEqual(result, "https://example.com/path?a=1&b=2") - - def test_exchange_ticker_hints_are_deduped(self): - hints = extract_exchange_ticker_hints( - "NVIDIA Corporation (NASDAQ: NVDA) and IBM NYSE: IBM. " - "Again (NASDAQ: NVDA)." - ) - - self.assertEqual([h.symbol for h in hints], ["NVDA", "IBM"]) - self.assertEqual(hints[0].exchange, "NASDAQ") - self.assertEqual(hints[1].exchange, "NYSE") - - def test_exchange_ticker_hints_ignore_markdown_decoration(self): - cases = ( - ( - "link", - "Carnival Corporation & plc " - "(NYSE: [CCL](#financial-modal)) today announced...", - ("CCL", "NYSE", "(NYSE: CCL)"), - ), - ( - "bold emphasis", - "Example Corporation (NASDAQ: **ABC**) today announced...", - ("ABC", "NASDAQ", "(NASDAQ: ABC)"), - ), - ( - "italic emphasis", - "Example Corporation (NASDAQ: *ABC*) today announced...", - ("ABC", "NASDAQ", "(NASDAQ: ABC)"), - ), - ) - - for name, text, expected in cases: - with self.subTest(name=name): - hints = extract_exchange_ticker_hints(text) - - self.assertEqual(len(hints), 1) - self.assertEqual( - (hints[0].symbol, hints[0].exchange, hints[0].raw), - expected, - ) - - def test_exchange_ticker_hints_capture_shared_prefix_comma_lists(self): - cases = ( - ( - "supported shared prefix", - "(NYSE: EVEX, EVEXW; B3: EVEB31)", - ( - ("EVEX", "NYSE", "(NYSE: EVEX"), - ("EVEXW", "NYSE", "NYSE: EVEXW"), - ), - ), - ("unsupported exchange", "(OTCID: QVCAQ, QVCGQ, QVCPQ)", ()), - ( - "conjunction boundary", - "(NYSE: TME and HKEX: 1698)", - (("TME", "NYSE", "(NYSE: TME"),), - ), - ( - "semicolon boundary TSXV", - "(NASDAQ: VMAR; TSXV: VMAR)", - (("VMAR", "NASDAQ", "(NASDAQ: VMAR"),), - ), - ( - "semicolon boundary BMV", - "(NYSE: ASR; BMV: ASUR)", - (("ASR", "NYSE", "(NYSE: ASR"),), - ), - ) - - for name, text, expected in cases: - with self.subTest(name=name): - hints = extract_exchange_ticker_hints(text) - - self.assertEqual( - tuple( - (hint.symbol, hint.exchange, hint.raw) for hint in hints - ), - expected, - ) - - def test_build_sec_news_identity(self): - self.assertEqual( - build_sec_news_identity( - accession_number="0001045810-26-000123", - section_key="EX99.1", - ), - "sec:0001045810-26-000123:ex99.1", - ) - - def test_build_news_identity_requires_source_reference(self): - with self.assertRaises(ValueError): - build_news_identity(source_type="press_release") - - def test_preprocess_news_document_builds_candidate_contract(self): - published = datetime( - 2024, 4, 15, 10, 30, tzinfo=timezone(timedelta(hours=-4)) - ) - raw = RawNewsDocument( - body_text=( - "NVIDIA Corporation (NASDAQ: NVDA) today reported " - "record quarterly revenue." - ), - source_url="https://example.com/pr/nvidia-results?utm_source=feed", - title="NVIDIA Announces Results", - publisher="PR Newswire", - published_at=published, - payload_id="gnw-123", - ) - - candidate = preprocess_news_document(raw) - - self.assertEqual(candidate.source_type, "press_release") - self.assertTrue(candidate.identity.startswith("wire:")) - self.assertEqual( - candidate.source_url, "https://example.com/pr/nvidia-results" - ) - self.assertEqual(candidate.published_at.tzinfo, timezone.utc) - self.assertEqual(candidate.published_at.hour, 14) - self.assertEqual(candidate.ticker_hints[0].symbol, "NVDA") - self.assertEqual( - candidate.content_hash, - hashlib.sha256(candidate.body_text.encode("utf-8")).hexdigest(), - ) - - def test_preprocess_news_document_rejects_empty_body(self): - raw = RawNewsDocument( - body_text=" ", - source_url="https://example.com/pr/empty", - ) - - with self.assertRaises(ValueError): - preprocess_news_document(raw) - - def test_wire_markup_fixture_meets_ticker_recall_control(self): - body_text = (_FIXTURES / "ticker_markup.md").read_text(encoding="utf-8") - raw = RawNewsDocument( - body_text=body_text, - source_url="https://www.prnewswire.com/news-releases/example.html", - ) - - candidate = preprocess_news_document(raw) - - expected_hints = { - ("ABC", "NASDAQ"), - ("CCL", "NYSE"), - ("PODD", "NASDAQ"), - } - actual_hints = { - (hint.symbol, hint.exchange) for hint in candidate.ticker_hints - } - recall = len(expected_hints & actual_hints) / len(expected_hints) - self.assertEqual(recall, 1.0) - self.assertIn("[CCL](#financial-modal)", candidate.body_text) - self.assertIn( - "(NASDAQ:\n\n[PODD](#financial-modal))", - candidate.body_text, - ) - self.assertIn("**ABC**", candidate.body_text) - self.assertEqual( - candidate.content_hash, - news_content_hash(candidate.body_text), - ) - - -class NewsFeedItemTests(unittest.IsolatedAsyncioTestCase): - async def test_feed_item_to_news_document_uses_inline_content(self): - item = FeedItem( - title="NVIDIA Announces Results", - url="https://example.com/pr/nvidia-results", - id="gnw-123", - published_at=datetime(2024, 4, 15, tzinfo=timezone.utc), - content_html=( - "

NVIDIA Corporation " - "(NASDAQ: NVDA) reported results.

" - ), - source_feed_url="https://example.com/rss", - ) - - raw = await feed_item_to_news_document(item, publisher="PR Newswire") - - self.assertEqual(raw.payload_id, "gnw-123") - self.assertEqual(raw.publisher, "PR Newswire") - self.assertIn("NVIDIA", raw.body_text) - self.assertEqual( - raw.metadata["source_feed_url"], "https://example.com/rss" - ) - self.assertEqual(raw.metadata["body_source"], "feed") - - async def test_article_source_fetches_despite_nonempty_teaser(self): - item = FeedItem( - title="NVIDIA Announces Results", - url="https://example.com/pr/nvidia-results", - id="wire-123", - summary_html="

This non-empty teaser is not the full body.

", - source_feed_url="https://example.com/rss", - ) - html = """ -
-

NVIDIA Announces Results

-

The complete release contains full financial details.

-
- """ - with respx.mock(assert_all_called=True) as router: - router.get(item.url).mock( - return_value=httpx.Response( - 200, - headers={"Content-Type": "text/html"}, - text=html, - ) - ) - raw = await feed_item_to_news_document( - item, - body_source="article", - ) - - self.assertIn("complete release", raw.body_text) - self.assertNotIn("non-empty teaser", raw.body_text) - self.assertEqual(raw.metadata["body_source"], "article") - - async def test_feed_source_does_not_fetch_item_url(self): - item = FeedItem( - title="NVIDIA Announces Results", - url="https://example.com/pr/nvidia-results", - summary_html="

Feed body stays local.

", - ) - - with respx.mock(assert_all_called=True): - raw = await feed_item_to_news_document( - item, - body_source="feed", - ) - - self.assertIn("Feed body", raw.body_text) - - async def test_article_source_requires_item_url(self): - item = FeedItem(title="Missing URL", summary_html="A teaser") - - with self.assertRaisesRegex(ValueError, "requires an item URL"): - await feed_item_to_news_document( - item, - body_source="article", - ) - - async def test_article_extraction_failure_is_explicit(self): - item = FeedItem( - title="Unsupported body", - url="https://example.com/release.pdf", - summary_html="A teaser", - ) - with respx.mock(assert_all_called=True) as router: - router.get(item.url).mock( - return_value=httpx.Response( - 200, - headers={"Content-Type": "application/pdf"}, - content=b"not html", - ) - ) - with self.assertRaisesRegex(ValueError, "Unsupported"): - await feed_item_to_news_document( - item, - body_source="article", - ) - - async def test_preprocess_feed_item_returns_candidate(self): - item = FeedItem( - title="NVIDIA Announces Results", - url="https://example.com/pr/nvidia-results", - id="gnw-123", - summary_html="

NVIDIA Corporation (NASDAQ: NVDA) reported results.

", - ) - - candidate = await preprocess_feed_item(item, publisher="PR Newswire") - - self.assertEqual(candidate.publisher, "PR Newswire") - self.assertEqual(candidate.ticker_hints[0].symbol, "NVDA") - - -class FetchNewsDocumentTests(unittest.IsolatedAsyncioTestCase): - async def test_fetch_news_document_extracts_html(self): - html = """ - - -
-

NVIDIA Announces Results

-

NVIDIA Corporation (NASDAQ: NVDA) reported record revenue.

-
- - - """ - with respx.mock(assert_all_called=True) as router: - router.get("https://example.com/pr/nvidia-results").mock( - return_value=httpx.Response( - 200, - headers={"Content-Type": "text/html; charset=utf-8"}, - content=html.encode(), - ) - ) - raw = await fetch_news_document( - "https://example.com/pr/nvidia-results", - title="NVIDIA Announces Results", - payload_id="gnw-123", - ) - - self.assertEqual( - raw.source_url, "https://example.com/pr/nvidia-results" - ) - self.assertEqual(raw.metadata["content_type"], "text/html") - self.assertIn("record revenue", raw.body_text) - - -if __name__ == "__main__": - unittest.main() +"""Tests for preprocess.news — PR/wire candidate normalisation.""" + +import hashlib +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import httpx +import respx + +from quantmind.preprocess.fetch.rss import FeedItem +from quantmind.preprocess.news import ( + RawNewsDocument, + build_news_identity, + build_sec_news_identity, + canonicalize_source_url, + extract_exchange_ticker_hints, + feed_item_to_news_document, + fetch_news_document, + news_content_hash, + normalize_news_text, + preprocess_feed_item, + preprocess_news_document, +) +from quantmind.preprocess.time import parse_news_datetime + +_FIXTURES = Path(__file__).parent / "fixtures" / "pr_newswire" + + +class NewsTimeTests(unittest.TestCase): + def test_parse_rfc822_news_datetime(self): + result = parse_news_datetime("Mon, 15 Apr 2024 10:30:00 GMT") + + self.assertEqual(result.tzinfo, timezone.utc) + self.assertEqual(result.year, 2024) + self.assertEqual(result.hour, 10) + + def test_parse_news_datetime_with_offset(self): + result = parse_news_datetime("Mon, 15 Apr 2024 10:30:00 -0400") + + self.assertEqual(result.tzinfo, timezone.utc) + self.assertEqual(result.hour, 14) + + +class NewsPreprocessTests(unittest.TestCase): + def test_normalize_news_text_and_hash(self): + text = ( + "NVIDIA\u2014reported revenue\n\n\nNVIDIA\u2014reported revenue" + ) + normalized = normalize_news_text(text) + + self.assertEqual( + normalized, + "NVIDIA-reported revenue\n\nNVIDIA-reported revenue", + ) + self.assertEqual( + news_content_hash(normalized), + hashlib.sha256(normalized.encode("utf-8")).hexdigest(), + ) + + def test_normalize_news_text_stabilizes_email_protection_links(self): + text_a = ( + "Contact [[email protected]](/cdn-cgi/l/email-protection#abc123)" + ) + text_b = ( + "Contact [[email protected]](/cdn-cgi/l/email-protection#def456)" + ) + + self.assertEqual( + normalize_news_text(text_a), normalize_news_text(text_b) + ) + + def test_canonicalize_source_url_drops_tracking(self): + result = canonicalize_source_url( + "HTTPS://Example.COM/path/?b=2&utm_source=x&a=1#frag" + ) + + self.assertEqual(result, "https://example.com/path?a=1&b=2") + + def test_exchange_ticker_hints_are_deduped(self): + hints = extract_exchange_ticker_hints( + "NVIDIA Corporation (NASDAQ: NVDA) and IBM NYSE: IBM. " + "Again (NASDAQ: NVDA)." + ) + + self.assertEqual([h.symbol for h in hints], ["NVDA", "IBM"]) + self.assertEqual(hints[0].exchange, "NASDAQ") + self.assertEqual(hints[1].exchange, "NYSE") + + def test_exchange_ticker_hints_ignore_markdown_decoration(self): + cases = ( + ( + "link", + "Carnival Corporation & plc " + "(NYSE: [CCL](#financial-modal)) today announced...", + ("CCL", "NYSE", "(NYSE: CCL)"), + ), + ( + "bold emphasis", + "Example Corporation (NASDAQ: **ABC**) today announced...", + ("ABC", "NASDAQ", "(NASDAQ: ABC)"), + ), + ( + "italic emphasis", + "Example Corporation (NASDAQ: *ABC*) today announced...", + ("ABC", "NASDAQ", "(NASDAQ: ABC)"), + ), + ) + + for name, text, expected in cases: + with self.subTest(name=name): + hints = extract_exchange_ticker_hints(text) + + self.assertEqual(len(hints), 1) + self.assertEqual( + (hints[0].symbol, hints[0].exchange, hints[0].raw), + expected, + ) + + def test_exchange_ticker_hints_capture_shared_prefix_comma_lists(self): + cases = ( + ( + "supported shared prefix", + "(NYSE: EVEX, EVEXW; B3: EVEB31)", + ( + ("EVEX", "NYSE", "(NYSE: EVEX"), + ("EVEXW", "NYSE", "NYSE: EVEXW"), + ), + ), + ("unsupported exchange", "(OTCID: QVCAQ, QVCGQ, QVCPQ)", ()), + ( + "conjunction boundary", + "(NYSE: TME and HKEX: 1698)", + (("TME", "NYSE", "(NYSE: TME"),), + ), + ( + "semicolon boundary TSXV", + "(NASDAQ: VMAR; TSXV: VMAR)", + (("VMAR", "NASDAQ", "(NASDAQ: VMAR"),), + ), + ( + "semicolon boundary BMV", + "(NYSE: ASR; BMV: ASUR)", + (("ASR", "NYSE", "(NYSE: ASR"),), + ), + ) + + for name, text, expected in cases: + with self.subTest(name=name): + hints = extract_exchange_ticker_hints(text) + + self.assertEqual( + tuple( + (hint.symbol, hint.exchange, hint.raw) for hint in hints + ), + expected, + ) + + def test_build_sec_news_identity(self): + self.assertEqual( + build_sec_news_identity( + accession_number="0001045810-26-000123", + section_key="EX99.1", + ), + "sec:0001045810-26-000123:ex99.1", + ) + + def test_build_news_identity_requires_source_reference(self): + with self.assertRaises(ValueError): + build_news_identity(source_type="press_release") + + def test_preprocess_news_document_builds_candidate_contract(self): + published = datetime( + 2024, 4, 15, 10, 30, tzinfo=timezone(timedelta(hours=-4)) + ) + raw = RawNewsDocument( + body_text=( + "NVIDIA Corporation (NASDAQ: NVDA) today reported " + "record quarterly revenue." + ), + source_url="https://example.com/pr/nvidia-results?utm_source=feed", + title="NVIDIA Announces Results", + publisher="PR Newswire", + published_at=published, + payload_id="gnw-123", + ) + + candidate = preprocess_news_document(raw) + + self.assertEqual(candidate.source_type, "press_release") + self.assertTrue(candidate.identity.startswith("wire:")) + self.assertEqual( + candidate.source_url, "https://example.com/pr/nvidia-results" + ) + self.assertEqual(candidate.published_at.tzinfo, timezone.utc) + self.assertEqual(candidate.published_at.hour, 14) + self.assertEqual(candidate.ticker_hints[0].symbol, "NVDA") + self.assertEqual( + candidate.content_hash, + hashlib.sha256(candidate.body_text.encode("utf-8")).hexdigest(), + ) + + def test_preprocess_news_document_rejects_empty_body(self): + raw = RawNewsDocument( + body_text=" ", + source_url="https://example.com/pr/empty", + ) + + with self.assertRaises(ValueError): + preprocess_news_document(raw) + + def test_wire_markup_fixture_meets_ticker_recall_control(self): + body_text = (_FIXTURES / "ticker_markup.md").read_text(encoding="utf-8") + raw = RawNewsDocument( + body_text=body_text, + source_url="https://www.prnewswire.com/news-releases/example.html", + ) + + candidate = preprocess_news_document(raw) + + expected_hints = { + ("ABC", "NASDAQ"), + ("CCL", "NYSE"), + ("PODD", "NASDAQ"), + } + actual_hints = { + (hint.symbol, hint.exchange) for hint in candidate.ticker_hints + } + recall = len(expected_hints & actual_hints) / len(expected_hints) + self.assertEqual(recall, 1.0) + self.assertIn("[CCL](#financial-modal)", candidate.body_text) + self.assertIn( + "(NASDAQ:\n\n[PODD](#financial-modal))", + candidate.body_text, + ) + self.assertIn("**ABC**", candidate.body_text) + self.assertEqual( + candidate.content_hash, + news_content_hash(candidate.body_text), + ) + + +class NewsFeedItemTests(unittest.IsolatedAsyncioTestCase): + async def test_feed_item_to_news_document_uses_inline_content(self): + item = FeedItem( + title="NVIDIA Announces Results", + url="https://example.com/pr/nvidia-results", + id="gnw-123", + published_at=datetime(2024, 4, 15, tzinfo=timezone.utc), + content_html=( + "

NVIDIA Corporation " + "(NASDAQ: NVDA) reported results.

" + ), + source_feed_url="https://example.com/rss", + ) + + raw = await feed_item_to_news_document(item, publisher="PR Newswire") + + self.assertEqual(raw.payload_id, "gnw-123") + self.assertEqual(raw.publisher, "PR Newswire") + self.assertIn("NVIDIA", raw.body_text) + self.assertEqual( + raw.metadata["source_feed_url"], "https://example.com/rss" + ) + self.assertEqual(raw.metadata["body_source"], "feed") + + async def test_article_source_fetches_despite_nonempty_teaser(self): + item = FeedItem( + title="NVIDIA Announces Results", + url="https://example.com/pr/nvidia-results", + id="wire-123", + summary_html="

This non-empty teaser is not the full body.

", + source_feed_url="https://example.com/rss", + ) + html = """ +
+

NVIDIA Announces Results

+

The complete release contains full financial details.

+
+ """ + with respx.mock(assert_all_called=True) as router: + router.get(item.url).mock( + return_value=httpx.Response( + 200, + headers={"Content-Type": "text/html"}, + text=html, + ) + ) + raw = await feed_item_to_news_document( + item, + body_source="article", + ) + + self.assertIn("complete release", raw.body_text) + self.assertNotIn("non-empty teaser", raw.body_text) + self.assertEqual(raw.metadata["body_source"], "article") + + async def test_feed_source_does_not_fetch_item_url(self): + item = FeedItem( + title="NVIDIA Announces Results", + url="https://example.com/pr/nvidia-results", + summary_html="

Feed body stays local.

", + ) + + with respx.mock(assert_all_called=True): + raw = await feed_item_to_news_document( + item, + body_source="feed", + ) + + self.assertIn("Feed body", raw.body_text) + + async def test_article_source_requires_item_url(self): + item = FeedItem(title="Missing URL", summary_html="A teaser") + + with self.assertRaisesRegex(ValueError, "requires an item URL"): + await feed_item_to_news_document( + item, + body_source="article", + ) + + async def test_article_extraction_failure_is_explicit(self): + item = FeedItem( + title="Unsupported body", + url="https://example.com/release.pdf", + summary_html="A teaser", + ) + with respx.mock(assert_all_called=True) as router: + router.get(item.url).mock( + return_value=httpx.Response( + 200, + headers={"Content-Type": "application/pdf"}, + content=b"not html", + ) + ) + with self.assertRaisesRegex(ValueError, "Unsupported"): + await feed_item_to_news_document( + item, + body_source="article", + ) + + async def test_preprocess_feed_item_returns_candidate(self): + item = FeedItem( + title="NVIDIA Announces Results", + url="https://example.com/pr/nvidia-results", + id="gnw-123", + summary_html="

NVIDIA Corporation (NASDAQ: NVDA) reported results.

", + ) + + candidate = await preprocess_feed_item(item, publisher="PR Newswire") + + self.assertEqual(candidate.publisher, "PR Newswire") + self.assertEqual(candidate.ticker_hints[0].symbol, "NVDA") + + +class FetchNewsDocumentTests(unittest.IsolatedAsyncioTestCase): + async def test_fetch_news_document_extracts_html(self): + html = """ + + +
+

NVIDIA Announces Results

+

NVIDIA Corporation (NASDAQ: NVDA) reported record revenue.

+
+ + + """ + with respx.mock(assert_all_called=True) as router: + router.get("https://example.com/pr/nvidia-results").mock( + return_value=httpx.Response( + 200, + headers={"Content-Type": "text/html; charset=utf-8"}, + content=html.encode(), + ) + ) + raw = await fetch_news_document( + "https://example.com/pr/nvidia-results", + title="NVIDIA Announces Results", + payload_id="gnw-123", + ) + + self.assertEqual( + raw.source_url, "https://example.com/pr/nvidia-results" + ) + self.assertEqual(raw.metadata["content_type"], "text/html") + self.assertIn("record revenue", raw.body_text) + + +if __name__ == "__main__": + unittest.main()