fix(preprocess): recover markdown-wrapped ticker hints

docs/README.md

@@ -28,9 +28,10 @@ layer shown in the catalog.
 |---|---|---|---|
 | PR Newswire | `NewsWindow(source="pr-newswire", ...)` | `collect_news` | `python scripts/verify_news_e2e.py` |
 
-The PR Newswire gate checks the public RSS feed and a complete preceding
-24-hour listing window without fetching article pages. Its GitHub workflow
-runs on pull requests, daily, and on manual dispatch.
+The PR Newswire gate checks the public RSS feed, a complete preceding 24-hour
+listing window, and ticker-hint recall on a bounded sample of up to 25 article
+pages. Its GitHub workflow runs on pull requests, daily, and on manual
+dispatch.
 
 ## Verification
 

docs/design/en/news.md

@@ -204,17 +204,22 @@ observations, raw-byte retention, retries, partial failures, and completeness.
 ### Live news E2E
 
 `python scripts/verify_news_e2e.py` is the canonical public-network component
-check. It performs two bounded checks:
+check. It performs three bounded checks:
 
 1. fetch and parse the official PR Newswire RSS feed;
 2. discover PR Newswire listing observations for the preceding 24 hours and
-   prove that discovery crossed the window start.
-
-The E2E check never fetches article pages. It prints component-level PASS/FAIL
-records and a compact observation/page/failure summary, then exits non-zero if
-RSS is empty or invalid, listing discovery is empty or incomplete, or a
-component raises. This keeps the check lightweight while detecting public
-source or parser drift.
+   prove that discovery crossed the window start;
+3. fetch up to 25 unique article observations and require 100% ticker-hint
+   recall for the first supported exchange-coded symbol in each group.
+
+The ticker control uses an oracle regex that is independent from the production
+extractor and accepts plain, Markdown-link, and emphasis-wrapped symbols. It
+intentionally ignores later members of multi-symbol lists and exchanges outside
+the production whitelist. The gate prints component-level PASS/FAIL records and
+compact discovery, article-sample, failure, and recall summaries. It exits
+non-zero if RSS or discovery is invalid, no sampled article parses, or a sample
+with supported exchange-coded symbols falls below 100% recall. A parsed sample
+with zero supported symbols reports `SKIP` and passes neutrally.
 
 GitHub Actions runs this check on every pull request, once daily, and on manual
 dispatch. The ordinary offline verification workflow remains network-free so

examples/preprocess/01_news_pr_wire.py

@@ -5,13 +5,13 @@
 from quantmind.preprocess.news import RawNewsDocument, preprocess_news_document
 
 raw = RawNewsDocument(
-    title="NVIDIA Announces Results",
+    title="Carnival Announces Results",
     body_text=(
-        "NVIDIA Corporation (NASDAQ: NVDA) today reported record quarterly "
-        "revenue and highlighted demand for accelerated computing."
+        "Carnival Corporation & plc (NYSE: [CCL](#financial-modal)) today "
+        "reported record quarterly revenue."
     ),
     source_type="press_release",
-    source_url="https://example.com/news/nvidia-results",
+    source_url="https://example.com/news/carnival-results",
     publisher="Example Newswire",
     published_at=datetime(2026, 7, 6, 12, 0, tzinfo=timezone.utc),
     payload_id="example-wire-123",
@@ -21,4 +21,5 @@
 
 print(candidate.identity)
 print(candidate.content_hash)
+print(candidate.body_text)
 print([hint.symbol for hint in candidate.ticker_hints])

quantmind/preprocess/news.py

@@ -51,6 +51,11 @@
     r"(?:\)|\b)",
     re.IGNORECASE,
 )
+_MARKDOWN_LINK_RE = re.compile(r"\[([^\]\n]+)\]\([^\)\n]*\)")
+_MARKDOWN_EMPHASIS_RE = re.compile(
+    r"(?<!\*)\*{1,2}([A-Z][A-Z0-9.-]{0,9})\*{1,2}(?!\*)",
+    re.IGNORECASE,
+)
 _EMAIL_PROTECTION_LINK_RE = re.compile(
     r"\[\[email protected]\]\(/cdn-cgi/l/email-protection#[^)]+\)"
 )
@@ -398,10 +403,14 @@ def extract_exchange_ticker_hints(text: str) -> tuple[NewsTickerHint, ...]:
 
     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(text):
+    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()

scripts/verify_news_e2e.py

@@ -2,28 +2,53 @@
 """Run the bounded live-network checks for public news collection."""
 
 import asyncio
+import re
 from datetime import datetime, timedelta, timezone
 
+from quantmind.preprocess import NewsDocument
 from quantmind.preprocess.fetch.http import FetchPolicy, HttpFetcher
 from quantmind.preprocess.fetch.rss import fetch_rss_feed
-from quantmind.preprocess.pr_newswire import _discover_pr_newswire
+from quantmind.preprocess.pr_newswire import (
+    PRNewswireDiscovery,
+    _collect_observation,
+    _discover_pr_newswire,
+)
 
 _PUBLIC_RSS_URL = "https://www.prnewswire.com/rss/news-releases-list.rss"
 _RSS_TIMEOUT_SECONDS = 60
 _DISCOVERY_TIMEOUT_SECONDS = 300
+_ARTICLE_SAMPLE_TIMEOUT_SECONDS = 180
+_ARTICLE_SAMPLE_SIZE = 25
+_LIVE_FETCH_POLICY = FetchPolicy(
+    max_attempts=3,
+    backoff_base_seconds=0.5,
+    backoff_max_seconds=5.0,
+    jitter_seconds=0.2,
+    max_concurrency_per_host=2,
+    min_interval_seconds=0.25,
+)
+
+# This oracle intentionally stays independent from the production extractor.
+# It checks the first supported symbol after each explicit exchange prefix. It
+# does not infer later list members or exchanges outside the current whitelist.
+_CONTROL_EXCHANGE_TICKER_RE = re.compile(
+    r"(?:\(|\b)"
+    r"(?:NASDAQ|NYSE(?:\s+American|\s+MKT|\s+Arca)?|"
+    r"AMEX|OTCQX|OTCQB|OTC|CBOE)"
+    r"\s*:\s*"
+    r"(?:"
+    r"\[(?P<link>[A-Z][A-Z0-9.-]{0,9})\]\([^\)\n]*\)"
+    r"|\*{1,2}(?P<emphasis>[A-Z][A-Z0-9.-]{0,9})\*{1,2}"
+    r"|(?P<plain>[A-Z][A-Z0-9.-]{0,9})"
+    r")",
+    re.IGNORECASE,
+)
 
 
 async def _check_rss() -> bool:
     try:
         async with HttpFetcher(
-            policy=FetchPolicy(
-                max_attempts=3,
-                backoff_base_seconds=0.5,
-                backoff_max_seconds=5.0,
-                jitter_seconds=0.2,
-                max_concurrency_per_host=1,
-                min_interval_seconds=0.25,
-            ),
+            policy=_LIVE_FETCH_POLICY,
             timeout=30.0,
             max_bytes=5_000_000,
         ) as fetcher:
@@ -48,15 +73,18 @@ async def _check_rss() -> bool:
     return passed
 
 
-async def _check_discovery(start: datetime, end: datetime) -> bool:
+async def _check_discovery(
+    start: datetime,
+    end: datetime,
+) -> tuple[bool, PRNewswireDiscovery | None]:
     try:
         result = await asyncio.wait_for(
             _discover_pr_newswire(start=start, end=end),
             timeout=_DISCOVERY_TIMEOUT_SECONDS,
         )
     except Exception as exc:
         print(f"[FAIL] pr-newswire-discovery: {type(exc).__name__}: {exc}")
-        return False
+        return False, None
 
     urls = [observation.canonical_url for observation in result.observations]
     duplicate_count = len(urls) - len(set(urls))
@@ -82,6 +110,90 @@ async def _check_discovery(start: datetime, end: datetime) -> bool:
         print(
             f"       {failure.stage} {failure.error_type} {failure.source_url}"
         )
+    return passed, result if passed else None
+
+
+def _ticker_hint_control_counts(
+    documents: tuple[NewsDocument, ...],
+) -> tuple[int, int]:
+    expected_count = 0
+    recovered_count = 0
+    for document in documents:
+        expected_symbols = {
+            (
+                match.group("link")
+                or match.group("emphasis")
+                or match.group("plain")
+            ).upper()
+            for match in _CONTROL_EXCHANGE_TICKER_RE.finditer(
+                document.cleaned_markdown
+            )
+        }
+        actual_symbols = {hint.symbol for hint in document.ticker_hints}
+        expected_count += len(expected_symbols)
+        recovered_count += len(expected_symbols & actual_symbols)
+    return expected_count, recovered_count
+
+
+async def _check_ticker_hints(discovery: PRNewswireDiscovery) -> bool:
+    sample = []
+    seen_urls: set[str] = set()
+    for observation in discovery.observations:
+        if observation.canonical_url in seen_urls:
+            continue
+        sample.append(observation)
+        seen_urls.add(observation.canonical_url)
+        if len(sample) == _ARTICLE_SAMPLE_SIZE:
+            break
+
+    if not sample:
+        print("[FAIL] pr-newswire-ticker-hints: no article URLs to sample")
+        return False
+
+    try:
+        async with HttpFetcher(
+            policy=_LIVE_FETCH_POLICY,
+            timeout=30.0,
+            max_bytes=10_000_000,
+        ) as fetcher:
+            outcomes = await asyncio.wait_for(
+                asyncio.gather(
+                    *(
+                        _collect_observation(
+                            observation,
+                            fetcher=fetcher,
+                            retain_raw_html=False,
+                        )
+                        for observation in sample
+                    )
+                ),
+                timeout=_ARTICLE_SAMPLE_TIMEOUT_SECONDS,
+            )
+    except Exception as exc:
+        print(f"[FAIL] pr-newswire-ticker-hints: {type(exc).__name__}: {exc}")
+        return False
+
+    documents = tuple(
+        outcome for outcome in outcomes if isinstance(outcome, NewsDocument)
+    )
+    expected_count, recovered_count = _ticker_hint_control_counts(documents)
+    article_failure_count = len(outcomes) - len(documents)
+    recall = recovered_count / expected_count if expected_count else 0.0
+    if not documents:
+        passed = False
+        state = "FAIL"
+    elif not expected_count:
+        passed = True
+        state = "SKIP"
+    else:
+        passed = recall == 1.0
+        state = "PASS" if passed else "FAIL"
+    print(
+        f"[{state}] pr-newswire-ticker-hints: sampled={len(sample)} "
+        f"parsed={len(documents)} failures={article_failure_count} "
+        f"expected={expected_count} recovered={recovered_count} "
+        f"recall={recall:.1%}"
+    )
     return passed
 
 
@@ -93,8 +205,13 @@ async def main(*, now: datetime | None = None) -> int:
     print(f"news window: [{start.isoformat()}, {end.isoformat()})")
 
     rss_passed = await _check_rss()
-    discovery_passed = await _check_discovery(start, end)
-    return 0 if rss_passed and discovery_passed else 1
+    discovery_passed, discovery = await _check_discovery(start, end)
+    if discovery is None:
+        print("[FAIL] pr-newswire-ticker-hints: discovery unavailable")
+        ticker_hints_passed = False
+    else:
+        ticker_hints_passed = await _check_ticker_hints(discovery)
+    return 0 if all((rss_passed, discovery_passed, ticker_hints_passed)) else 1
 
 
 if __name__ == "__main__":

tests/preprocess/fixtures/pr_newswire/ticker_markup.md

@@ -0,0 +1,9 @@
+# Wire ticker markup regression fixture
+
+Carnival Corporation & plc (NYSE: [CCL](#financial-modal)) today announced an update.
+
+Insulet Corporation (NASDAQ:
+
+[PODD](#financial-modal)) today announced an update.
+
+Example Corporation (NASDAQ: **ABC**) today announced an update.

tests/preprocess/test_news.py

@@ -3,6 +3,7 @@
 import hashlib
 import unittest
 from datetime import datetime, timedelta, timezone
+from pathlib import Path
 
 import httpx
 import respx
@@ -23,6 +24,8 @@
 )
 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):
@@ -84,6 +87,36 @@ def test_exchange_ticker_hints_are_deduped(self):
         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(
@@ -137,6 +170,36 @@ def test_preprocess_news_document_rejects_empty_body(self):
         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):