diff --git a/.agents/skills/quantmind-dev/SKILL.md b/.agents/skills/quantmind-dev/SKILL.md index a05338c..4b0cda8 100644 --- a/.agents/skills/quantmind-dev/SKILL.md +++ b/.agents/skills/quantmind-dev/SKILL.md @@ -11,7 +11,9 @@ Development workflow for contributing to the QuantMind codebase. 1. Read the repository root `AGENTS.md` or `CLAUDE.md` for the stable architecture constraints and the module map. -2. Pick exactly one workflow reference below; do not load the others. +2. Read `docs/README.md` when the task adds, changes, or uses a public + operation or public-network source. +3. Pick exactly one workflow reference below; do not load the others. ## Select Workflow @@ -24,8 +26,11 @@ A feature task usually chains all three: develop → commit → pull request. ## Rules -- `bash scripts/verify.sh` is the single "shippable" gate. CI runs the same - script; run it before every push and before marking a PR ready. +- `bash scripts/verify.sh` is the deterministic offline golden gate. Run it + before every push and before marking a PR ready. +- Public-network integrations have separate bounded live component gates. + Run every applicable gate listed in `docs/README.md` when changing that + component and before publishing. - Never bypass pre-commit / pre-push hooks (`--no-verify`) unless the user explicitly authorizes it. - New features ship with a unit test and a focused example (see diff --git a/.agents/skills/quantmind-dev/references/develop-components.md b/.agents/skills/quantmind-dev/references/develop-components.md index b04a647..88ddf66 100644 --- a/.agents/skills/quantmind-dev/references/develop-components.md +++ b/.agents/skills/quantmind-dev/references/develop-components.md @@ -18,10 +18,12 @@ apply throughout. or a side effect beyond the call it wraps). No premature abstractions — extract shared code when the second real caller appears, not before. 4. **Add the unit test and the example** (sections below). -5. **Update the public surface** if needed: module `__init__.py` exports - and user-facing docs (`README.md`, `docs/`) for user-visible features. -6. **Verify**: targeted `pytest tests//` while iterating; - `bash scripts/verify.sh` before handoff. +5. **Update the public surface** if needed: package exports, the relevant + design or guide, and the catalog in `docs/README.md`. Update the root + README only when its overview or quick start changes. +6. **Verify**: run targeted tests while iterating, the offline golden gate + before handoff, and every applicable live component gate for changed + public-network integrations. ## Module Routing @@ -46,9 +48,9 @@ apply throughout. objects are `TreeKnowledge` even when a flatten card exists alongside (e.g. `Paper` vs `PaperKnowledgeCard`). -### `quantmind/configs/` — flow cfg + typed inputs +### `quantmind/configs/` — operation cfg + typed inputs -- Extend `BaseFlowCfg`; inputs are discriminated-union Pydantic types. +- Extend `BaseFlowCfg`; inputs are Pydantic models or discriminated unions. - Never `Dict[str, Any]` in signatures — model it. ### `quantmind/preprocess/` — deterministic data prep @@ -60,10 +62,11 @@ apply throughout. ### `quantmind/flows/` and `quantmind/magic.py` — apex layer -- Flows are pure `async def` functions, not classes; state passes as - arguments; side effects go through explicit hooks. -- Use the OpenAI Agents SDK directly (`Agent`, `@function_tool`, - `output_type=`); never wrap `from agents import ...` in a facade. +- Public operations are `async def` functions, not classes; state passes + as arguments and side effects are explicit. +- Semantic operations use the OpenAI Agents SDK directly (`Agent`, + `@function_tool`, `output_type=`); deterministic operations do not add an + LLM. Never wrap `from agents import ...` in a facade. - Fan-out goes through `batch_run` (bounded concurrency, error policy); `batch_run` rejects `memory=` at the signature layer by design. @@ -79,6 +82,35 @@ apply throughout. - Logger only. New general-purpose helpers need maintainer sign-off via an issue first; the default answer is "put it in the module that uses it". +## Public Operation Checklist + +A public operation is complete only when all of these agree: + +1. A stage and name consistent with `docs/design/en/operations.md`. +2. Typed input and config models, exported from `quantmind.configs`. +3. One intent-oriented `async def` operation exported from `quantmind.flows`, + with its result contract exported from the canonical owning layer. +4. Offline success and failure tests plus a magic-introspection test when the + operation follows the `(input, *, cfg)` convention. +5. One runnable common-path example under `examples//`. +6. A relevant design or guide and one row in `docs/README.md`. + +Do not add a registry solely for discovery; package exports and the component +catalog are the discovery surfaces. + +## Public-Network Source Checklist + +When adding a source to an existing operation: + +1. Update the typed source selection and the operation's direct dispatch. +2. Keep acquisition policy internal. Add a shared provider abstraction only + after a second implementation demonstrates shared behavior. +3. Add offline mocked tests for parsing, boundaries, continuation after item + failures, and completeness semantics. +4. Update the source table and design under `docs/`. +5. Add or update a bounded live verifier and GitHub workflow. Keep live + network work out of `scripts/verify.sh`. + ## Tests - Location mirrors the module: `tests//test_.py`. @@ -101,5 +133,6 @@ apply throughout. - Docstrings: English, Google style, required on public functions and models. -- User-visible behavior changes update `README.md` (usage section) and - `docs/` where applicable. +- Public behavior changes update the relevant design or guide and the catalog + in `docs/README.md`. Update the root README only for top-level positioning + or quick-start changes. diff --git a/.agents/skills/quantmind-dev/references/pull-request.md b/.agents/skills/quantmind-dev/references/pull-request.md index 8f64552..9c79ac1 100644 --- a/.agents/skills/quantmind-dev/references/pull-request.md +++ b/.agents/skills/quantmind-dev/references/pull-request.md @@ -7,14 +7,16 @@ How to open and maintain a pull request against QuantMind. 1. Branch from `master`; keep the PR small and focused — one PR equals one reviewable change (see [Google eng practices on small CLs](https://google.github.io/eng-practices/review/developer/small-cls.html)). -2. Run the canonical gate and make sure it is green: +2. Run the deterministic offline golden gate: ```bash bash scripts/verify.sh ``` - CI runs the exact same script; do not open (or mark ready) a PR with a - red local run. +3. If the change affects a public-network integration, run every applicable + live component gate listed in `docs/README.md`. + +Do not open or mark ready a PR with a red offline or applicable live gate. ## Title @@ -37,9 +39,10 @@ and make sure the body covers: if one exists. 2. **Related issue** — reference it when one exists (`Closes #NN` / `Part of #NN`). -3. **Verification performed** — state what you ran (e.g. - `bash scripts/verify.sh` green; targeted `pytest tests//`; - manual example run) so the reviewer does not have to guess. +3. **Verification performed** — state the exact offline and applicable live + commands you ran (e.g. `bash scripts/verify.sh`; targeted + `pytest tests//`; a component verifier; manual example run) so the + reviewer does not have to guess. Keep the template checklist and remove items that do not apply. diff --git a/.claude/skills/quantmind-dev/SKILL.md b/.claude/skills/quantmind-dev/SKILL.md index a05338c..4b0cda8 100644 --- a/.claude/skills/quantmind-dev/SKILL.md +++ b/.claude/skills/quantmind-dev/SKILL.md @@ -11,7 +11,9 @@ Development workflow for contributing to the QuantMind codebase. 1. Read the repository root `AGENTS.md` or `CLAUDE.md` for the stable architecture constraints and the module map. -2. Pick exactly one workflow reference below; do not load the others. +2. Read `docs/README.md` when the task adds, changes, or uses a public + operation or public-network source. +3. Pick exactly one workflow reference below; do not load the others. ## Select Workflow @@ -24,8 +26,11 @@ A feature task usually chains all three: develop → commit → pull request. ## Rules -- `bash scripts/verify.sh` is the single "shippable" gate. CI runs the same - script; run it before every push and before marking a PR ready. +- `bash scripts/verify.sh` is the deterministic offline golden gate. Run it + before every push and before marking a PR ready. +- Public-network integrations have separate bounded live component gates. + Run every applicable gate listed in `docs/README.md` when changing that + component and before publishing. - Never bypass pre-commit / pre-push hooks (`--no-verify`) unless the user explicitly authorizes it. - New features ship with a unit test and a focused example (see diff --git a/.claude/skills/quantmind-dev/references/develop-components.md b/.claude/skills/quantmind-dev/references/develop-components.md index b04a647..88ddf66 100644 --- a/.claude/skills/quantmind-dev/references/develop-components.md +++ b/.claude/skills/quantmind-dev/references/develop-components.md @@ -18,10 +18,12 @@ apply throughout. or a side effect beyond the call it wraps). No premature abstractions — extract shared code when the second real caller appears, not before. 4. **Add the unit test and the example** (sections below). -5. **Update the public surface** if needed: module `__init__.py` exports - and user-facing docs (`README.md`, `docs/`) for user-visible features. -6. **Verify**: targeted `pytest tests//` while iterating; - `bash scripts/verify.sh` before handoff. +5. **Update the public surface** if needed: package exports, the relevant + design or guide, and the catalog in `docs/README.md`. Update the root + README only when its overview or quick start changes. +6. **Verify**: run targeted tests while iterating, the offline golden gate + before handoff, and every applicable live component gate for changed + public-network integrations. ## Module Routing @@ -46,9 +48,9 @@ apply throughout. objects are `TreeKnowledge` even when a flatten card exists alongside (e.g. `Paper` vs `PaperKnowledgeCard`). -### `quantmind/configs/` — flow cfg + typed inputs +### `quantmind/configs/` — operation cfg + typed inputs -- Extend `BaseFlowCfg`; inputs are discriminated-union Pydantic types. +- Extend `BaseFlowCfg`; inputs are Pydantic models or discriminated unions. - Never `Dict[str, Any]` in signatures — model it. ### `quantmind/preprocess/` — deterministic data prep @@ -60,10 +62,11 @@ apply throughout. ### `quantmind/flows/` and `quantmind/magic.py` — apex layer -- Flows are pure `async def` functions, not classes; state passes as - arguments; side effects go through explicit hooks. -- Use the OpenAI Agents SDK directly (`Agent`, `@function_tool`, - `output_type=`); never wrap `from agents import ...` in a facade. +- Public operations are `async def` functions, not classes; state passes + as arguments and side effects are explicit. +- Semantic operations use the OpenAI Agents SDK directly (`Agent`, + `@function_tool`, `output_type=`); deterministic operations do not add an + LLM. Never wrap `from agents import ...` in a facade. - Fan-out goes through `batch_run` (bounded concurrency, error policy); `batch_run` rejects `memory=` at the signature layer by design. @@ -79,6 +82,35 @@ apply throughout. - Logger only. New general-purpose helpers need maintainer sign-off via an issue first; the default answer is "put it in the module that uses it". +## Public Operation Checklist + +A public operation is complete only when all of these agree: + +1. A stage and name consistent with `docs/design/en/operations.md`. +2. Typed input and config models, exported from `quantmind.configs`. +3. One intent-oriented `async def` operation exported from `quantmind.flows`, + with its result contract exported from the canonical owning layer. +4. Offline success and failure tests plus a magic-introspection test when the + operation follows the `(input, *, cfg)` convention. +5. One runnable common-path example under `examples//`. +6. A relevant design or guide and one row in `docs/README.md`. + +Do not add a registry solely for discovery; package exports and the component +catalog are the discovery surfaces. + +## Public-Network Source Checklist + +When adding a source to an existing operation: + +1. Update the typed source selection and the operation's direct dispatch. +2. Keep acquisition policy internal. Add a shared provider abstraction only + after a second implementation demonstrates shared behavior. +3. Add offline mocked tests for parsing, boundaries, continuation after item + failures, and completeness semantics. +4. Update the source table and design under `docs/`. +5. Add or update a bounded live verifier and GitHub workflow. Keep live + network work out of `scripts/verify.sh`. + ## Tests - Location mirrors the module: `tests//test_.py`. @@ -101,5 +133,6 @@ apply throughout. - Docstrings: English, Google style, required on public functions and models. -- User-visible behavior changes update `README.md` (usage section) and - `docs/` where applicable. +- Public behavior changes update the relevant design or guide and the catalog + in `docs/README.md`. Update the root README only for top-level positioning + or quick-start changes. diff --git a/.claude/skills/quantmind-dev/references/pull-request.md b/.claude/skills/quantmind-dev/references/pull-request.md index 8f64552..9c79ac1 100644 --- a/.claude/skills/quantmind-dev/references/pull-request.md +++ b/.claude/skills/quantmind-dev/references/pull-request.md @@ -7,14 +7,16 @@ How to open and maintain a pull request against QuantMind. 1. Branch from `master`; keep the PR small and focused — one PR equals one reviewable change (see [Google eng practices on small CLs](https://google.github.io/eng-practices/review/developer/small-cls.html)). -2. Run the canonical gate and make sure it is green: +2. Run the deterministic offline golden gate: ```bash bash scripts/verify.sh ``` - CI runs the exact same script; do not open (or mark ready) a PR with a - red local run. +3. If the change affects a public-network integration, run every applicable + live component gate listed in `docs/README.md`. + +Do not open or mark ready a PR with a red offline or applicable live gate. ## Title @@ -37,9 +39,10 @@ and make sure the body covers: if one exists. 2. **Related issue** — reference it when one exists (`Closes #NN` / `Part of #NN`). -3. **Verification performed** — state what you ran (e.g. - `bash scripts/verify.sh` green; targeted `pytest tests//`; - manual example run) so the reviewer does not have to guess. +3. **Verification performed** — state the exact offline and applicable live + commands you ran (e.g. `bash scripts/verify.sh`; targeted + `pytest tests//`; a component verifier; manual example run) so the + reviewer does not have to guess. Keep the template checklist and remove items that do not apply. diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md index b21e545..a6e8f58 100644 --- a/.github/PULL_REQUEST_TEMPLATE.md +++ b/.github/PULL_REQUEST_TEMPLATE.md @@ -1,14 +1,20 @@ -## Description +## Summary - + -## Checklist +## Related Issue + + + +## Verification -Please feel free to remove inapplicable items for your PR. + + +## Checklist -- [ ] The PR title starts with `$CATEGORY(xx): xxx` (such as `feat(tool): xxx`, `fix(source): xxx`, `docs(README): xxx`) -- [ ] Related issue is referred in this PR -- [ ] The markdown and latex are rendered correctly. -- [ ] The code in PR is well-documented. -- [ ] The PR is complete and small, read the [Google eng practice (CL equals to PR)](https://google.github.io/eng-practices/review/developer/small-cls.html) to understand more about small PR. +- [ ] The title uses English Conventional Commit format: `type(scope): summary`. +- [ ] The related issue or design discussion is linked when applicable. +- [ ] `bash scripts/verify.sh` passes. +- [ ] Every applicable live component gate passes, or this PR states why none applies. +- [ ] Public behavior has focused tests, an example, and documentation where applicable. +- [ ] The PR is complete, small, and contains no unrelated changes. diff --git a/.github/workflows/news-e2e.yml b/.github/workflows/news-e2e.yml new file mode 100644 index 0000000..567db07 --- /dev/null +++ b/.github/workflows/news-e2e.yml @@ -0,0 +1,44 @@ +name: news-e2e + +on: + pull_request: + branches: [master] + schedule: + - cron: "17 3 * * *" + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + news-e2e: + runs-on: ubuntu-latest + timeout-minutes: 10 + + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.10' + + - name: Install uv + uses: astral-sh/setup-uv@v3 + with: + enable-cache: true + cache-dependency-glob: pyproject.toml + + - name: Create virtual environment + run: uv venv + + - name: Install project runtime dependencies + run: uv pip install --python .venv/bin/python -e . + + - name: Run live news E2E + run: .venv/bin/python scripts/verify_news_e2e.py diff --git a/AGENTS.md b/AGENTS.md index 4a7e302..d1b0d79 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -15,9 +15,9 @@ handoff all come from `openai-agents`. | Module | Role | |--------|------| | `quantmind/knowledge/` | Pydantic data standard (`FlattenKnowledge` / `TreeKnowledge` / `GraphKnowledge`) — dependency leaf | -| `quantmind/configs/` | Flow cfg + typed inputs (`BaseFlowCfg`, discriminated unions) — depends only on `knowledge` | +| `quantmind/configs/` | Operation cfg + typed input models or unions (`BaseFlowCfg`, `NewsWindow`, `PaperInput`) — depends only on `knowledge` | | `quantmind/preprocess/` | Deterministic fetch / format / clean / time utilities — depends only on `utils` | -| `quantmind/flows/` | Apex layer: end-to-end pipeline functions (`paper_flow`, `batch_run`) | +| `quantmind/flows/` | Apex layer: public library operations (`paper_flow`, `collect_news`, `batch_run`) | | `quantmind/magic.py` | `resolve_magic_input`: natural language → `(input, cfg)` | | `quantmind/mind/` | Cognitive layer (memory protocol); landing via the Agents SDK migration (#71) | | `quantmind/utils/` | Logger only — keep it that way | @@ -31,14 +31,17 @@ resurrect it into master. ```bash uv venv && source .venv/bin/activate uv pip install -e ".[dev]" -bash scripts/verify.sh # canonical "is this branch shippable" check +bash scripts/verify.sh # deterministic offline golden gate +python scripts/verify_news_e2e.py # live PR Newswire component gate ``` `scripts/verify.sh` runs five fast-fail steps (`ruff format --check`, -`ruff check`, `basedpyright`, `lint-imports`, `pytest --cov`). CI runs the -exact same script, so a green local run means a green PR. Do not bypass -pre-commit / pre-push hooks unless the user explicitly authorizes it — fix -the underlying issue instead. +`ruff check`, `basedpyright`, `lint-imports`, `pytest --cov`) and must remain +network-free. Public-network integrations have separate bounded live gates; +run each applicable gate when changing that component and before publishing. +The current component catalog and commands live in `docs/README.md`. Do not +bypass pre-commit / pre-push hooks unless the user explicitly authorizes it — +fix the underlying issue instead. ## Architecture Constraints (stable) @@ -46,15 +49,19 @@ the underlying issue instead. no plugin registries, no hook discovery, no CLI. 2. **Do not rebuild the agent runtime** — use `openai-agents` directly; no QuantMind-side facades over `from agents import ...`. -3. **Pydantic at boundaries, frozen dataclass internally** — Pydantic - (`frozen=True`, `extra="forbid"`) for anything exposed to an LLM or a - user; frozen dataclasses for internal value types. +3. **Schema models vs runtime evidence** — user/LLM inputs and configs use + extra-forbid Pydantic models; knowledge adds `frozen=True`; deterministic + fetch, preprocessing, and collection values use frozen dataclasses when + they do not need validation or JSON Schema (`Fetched`, `NewsBatch`). 4. **Import boundaries are contracts** — `import-linter` (configured in `pyproject.toml`) pins the dependency graph; never work around a failing contract. 5. **Absolute imports** across module boundaries. 6. **No meaningless wrappers** — a method must add logic, abstraction, or a side effect beyond the call it wraps; otherwise inline it. +7. **Name public operations by intent** — follow + `docs/design/en/operations.md`; use stage verbs, and reserve `pipeline` for + deliberate multi-stage composition. ## Tests and Examples @@ -63,6 +70,8 @@ A new feature ships with a unit test **and** a focused example: - Tests: `tests//`, subclass `unittest.TestCase`, mock external services, cover success and failure paths. - Examples: `examples//`, one simple usage per file. +- Public operations and sources: update the catalog in `docs/README.md` and + follow the `quantmind-dev` component checklist. ## Communication diff --git a/CLAUDE.md b/CLAUDE.md index 477b577..b5c66d2 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -16,9 +16,9 @@ handoff all come from `openai-agents`. | Module | Role | |--------|------| | `quantmind/knowledge/` | Pydantic data standard (`FlattenKnowledge` / `TreeKnowledge` / `GraphKnowledge`) — dependency leaf | -| `quantmind/configs/` | Flow cfg + typed inputs (`BaseFlowCfg`, discriminated unions) — depends only on `knowledge` | +| `quantmind/configs/` | Operation cfg + typed input models or unions (`BaseFlowCfg`, `NewsWindow`, `PaperInput`) — depends only on `knowledge` | | `quantmind/preprocess/` | Deterministic fetch / format / clean / time utilities — depends only on `utils` | -| `quantmind/flows/` | Apex layer: end-to-end pipeline functions (`paper_flow`, `batch_run`) | +| `quantmind/flows/` | Apex layer: public library operations (`paper_flow`, `collect_news`, `batch_run`) | | `quantmind/magic.py` | `resolve_magic_input`: natural language → `(input, cfg)` | | `quantmind/mind/` | Cognitive layer (memory protocol); landing via the Agents SDK migration (#71) | | `quantmind/utils/` | Logger only — keep it that way | @@ -32,14 +32,17 @@ resurrect it into master. ```bash uv venv && source .venv/bin/activate uv pip install -e ".[dev]" -bash scripts/verify.sh # canonical "is this branch shippable" check +bash scripts/verify.sh # deterministic offline golden gate +python scripts/verify_news_e2e.py # live PR Newswire component gate ``` `scripts/verify.sh` runs five fast-fail steps (`ruff format --check`, -`ruff check`, `basedpyright`, `lint-imports`, `pytest --cov`). CI runs the -exact same script, so a green local run means a green PR. Do not bypass -pre-commit / pre-push hooks unless the user explicitly authorizes it — fix -the underlying issue instead. +`ruff check`, `basedpyright`, `lint-imports`, `pytest --cov`) and must remain +network-free. Public-network integrations have separate bounded live gates; +run each applicable gate when changing that component and before publishing. +The current component catalog and commands live in `docs/README.md`. Do not +bypass pre-commit / pre-push hooks unless the user explicitly authorizes it — +fix the underlying issue instead. ## Architecture Constraints (stable) @@ -47,15 +50,19 @@ the underlying issue instead. no plugin registries, no hook discovery, no CLI. 2. **Do not rebuild the agent runtime** — use `openai-agents` directly; no QuantMind-side facades over `from agents import ...`. -3. **Pydantic at boundaries, frozen dataclass internally** — Pydantic - (`frozen=True`, `extra="forbid"`) for anything exposed to an LLM or a - user; frozen dataclasses for internal value types. +3. **Schema models vs runtime evidence** — user/LLM inputs and configs use + extra-forbid Pydantic models; knowledge adds `frozen=True`; deterministic + fetch, preprocessing, and collection values use frozen dataclasses when + they do not need validation or JSON Schema (`Fetched`, `NewsBatch`). 4. **Import boundaries are contracts** — `import-linter` (configured in `pyproject.toml`) pins the dependency graph; never work around a failing contract. 5. **Absolute imports** across module boundaries. 6. **No meaningless wrappers** — a method must add logic, abstraction, or a side effect beyond the call it wraps; otherwise inline it. +7. **Name public operations by intent** — follow + `docs/design/en/operations.md`; use stage verbs, and reserve `pipeline` for + deliberate multi-stage composition. ## Tests and Examples @@ -64,6 +71,8 @@ A new feature ships with a unit test **and** a focused example: - Tests: `tests//`, subclass `unittest.TestCase`, mock external services, cover success and failure paths. - Examples: `examples//`, one simple usage per file. +- Public operations and sources: update the catalog in `docs/README.md` and + follow the `quantmind-dev` component checklist. ## Communication diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index e5eb4d7..c3e787a 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -26,9 +26,7 @@ will pick these up automatically. ## ✅ Verification -`scripts/verify.sh` is the single source of truth for "is this branch -shippable". CI runs the exact same script, so a green local run means a -green PR: +`scripts/verify.sh` is the deterministic offline golden gate for every PR: ```bash bash scripts/verify.sh @@ -36,7 +34,16 @@ bash scripts/verify.sh It runs five fast-fail steps: `ruff format --check`, `ruff check`, `basedpyright`, `lint-imports`, `pytest --cov` (with a branch-coverage -floor configured in `pyproject.toml`). +floor configured in `pyproject.toml`). It stays network-free. + +Public-network integrations have separate bounded live component gates. +Run every applicable gate when changing that component and before publishing; +the current commands are listed in [`docs/README.md`](docs/README.md). For +example, PR Newswire uses: + +```bash +python scripts/verify_news_e2e.py +``` **Hooks**: the pre-commit stage runs formatting/lint and file hygiene checks; the pre-push stage runs the full `scripts/verify.sh`. If a hook @@ -58,8 +65,9 @@ pytest tests// constraints and the module map. - **Style**: Google-style docstrings, English comments, 80-char lines (enforced by ruff). -- **Types**: Pydantic models at boundaries, frozen dataclasses internally; - comprehensive type hints (`basedpyright` runs in standard mode). +- **Types**: Pydantic for inputs, configs, and knowledge schemas; frozen + dataclasses for deterministic runtime evidence; comprehensive type hints + (`basedpyright` runs in standard mode). - **Dependency boundaries**: enforced by `import-linter` (`pyproject.toml`); don't work around a failing contract. - **Tests**: required under `tests//` (mirror the module @@ -67,12 +75,15 @@ pytest tests// and failure paths. - **Examples**: one focused example under `examples//` for each new feature. +- **Public operations and sources**: update the component catalog in + `docs/README.md`; public-network sources also require a bounded live check. ## 🔄 Pull Request Process 1. **Create a feature branch** from `master`. 2. **Follow Conventional Commits**: `type(scope): description`, in English. -3. **Verify before submitting**: `bash scripts/verify.sh` must be green. +3. **Verify before submitting**: the offline golden gate and every applicable + live component gate must be green. 4. **Submit the PR** using the template — English body, reference the related issue, and state the verification you performed. 5. Keep PRs small and focused diff --git a/docs/NEWS_PIPELINE.md b/docs/NEWS_PIPELINE.md deleted file mode 100644 index 2302489..0000000 --- a/docs/NEWS_PIPELINE.md +++ /dev/null @@ -1,247 +0,0 @@ -# News Pipeline - -QuantMind provides deterministic building blocks for fetching, parsing, and -normalizing public financial news. The current wire-provider MVP supports PR -Newswire. The provider contract is extensible, but other wire services are not -built-in or verified yet. - -This guide describes the public OSS behavior implemented today. It does not -assume private feeds, credentials, persistence infrastructure, or internal -services. - -## What the Pipeline Can Do - -| Stage | Public API | Result | -| --- | --- | --- | -| HTTP acquisition | `fetch_url`, `HttpFetcher`, `FetchPolicy` | Bounded response bytes and fetch metadata | -| Feed parsing | `fetch_rss_feed`, `parse_feed` | Typed RSS 2.x or Atom items | -| Article acquisition | `feed_item_to_news_document` | Explicit feed-body or linked-article text | -| Normalization | `preprocess_news_document`, `preprocess_feed_item` | Deterministic `NewsCandidate` | -| Wire ingestion | `fetch_wire_documents` | PR Newswire documents plus non-fatal failures | - -The main wire path is: - -```text -Public RSS/Atom feed - -> shared HTTP fetcher - -> typed feed items - -> provider mapping - -> linked article fetch - -> deterministic news normalization - -> WireDocument + optional WireFetchFailure records -``` - -### HTTP robustness - -`FetchPolicy` can opt a caller into: - -- bounded retries for transport errors, HTTP 408, HTTP 429, and HTTP 5xx; -- exponential backoff with configurable jitter; -- `Retry-After` handling for HTTP 429 and HTTP 503; -- per-host concurrency limits and minimum request spacing. - -Calling `fetch_url` without a policy preserves its one-shot behavior. Reuse one -`HttpFetcher` when several requests should share a connection pool and the same -per-host rate state. - -### RSS and Atom parsing - -`fetch_rss_feed` and `parse_feed` support RSS 2.x and Atom. Each `FeedItem` -retains normalized fields, selected source metadata, and the raw XML entry. - -### Explicit body acquisition - -News feed items use an explicit body-source decision: - -- `body_source="feed"` trusts the content included in the feed and performs no - article request; -- `body_source="article"` always follows the item URL, even when the feed - contains a non-empty teaser. - -The default is `feed`. The built-in PR Newswire adapter uses `article` because -its public feed descriptions may be teasers. - -### Deterministic normalization - -`NewsCandidate` normalization currently includes: - -- Unicode, whitespace, and duplicate-line cleanup; -- canonical source URLs with common tracking parameters removed; -- UTC publication timestamps; -- stable content hashes and source deduplication keys; -- exchange-qualified ticker hints such as `NASDAQ: NVDA` or `NYSE: IBM`. - -Ticker extraction only produces hints. Instrument resolution and validation -remain downstream responsibilities. - -### Replayable wire documents - -Each successful `WireDocument` contains: - -- the raw feed entry; -- raw article HTML when an article request was used; -- source and fetch metadata; -- cleaned Markdown; -- provider identity, payload ID, canonical URL, title, publisher, publication - time, content hash, stable identity, and ticker hints. - -Individual feed or article failures do not fail the whole call. They are -returned as lightweight `WireFetchFailure` values alongside successful -documents. - -## Examples - -### Normalize an existing news document - -This path performs no network request and is useful when the caller already -has the article text. - -```python -from datetime import datetime, timezone - -from quantmind.preprocess import RawNewsDocument, preprocess_news_document - -raw = RawNewsDocument( - title="Example Corp Reports Results", - body_text=( - "Example Corp (NASDAQ: EXMPL) reported fictional quarterly revenue." - ), - source_url="https://example.com/releases/results?utm_source=rss", - publisher="Example Publisher", - published_at=datetime(2026, 7, 12, 12, 0, tzinfo=timezone.utc), - payload_id="release-123", -) - -candidate = preprocess_news_document(raw) - -print(candidate.source_url) -print(candidate.dedup_key) -print([hint.symbol for hint in candidate.ticker_hints]) -``` - -A runnable version lives at -[`examples/preprocess/01_news_pr_wire.py`](../examples/preprocess/01_news_pr_wire.py). - -### Fetch and normalize a custom RSS or Atom feed - -Choose the body source based on the feed contract. Use `article` when entries -only contain teasers. - -```python -import asyncio - -from quantmind.preprocess import ( - FetchPolicy, - HttpFetcher, - fetch_rss_feed, - preprocess_feed_item, -) - - -async def main() -> None: - policy = FetchPolicy( - max_attempts=3, - max_concurrency_per_host=2, - min_interval_seconds=0.25, - ) - async with HttpFetcher(policy=policy) as fetcher: - feed = await fetch_rss_feed( - "https://example.com/news.xml", - fetcher=fetcher, - ) - candidates = [ - await preprocess_feed_item( - item, - publisher="Example Publisher", - body_source="article", - fetcher=fetcher, - ) - for item in feed.items - ] - - print(f"candidates={len(candidates)}") - - -asyncio.run(main()) -``` - -### Fetch the public PR Newswire feed - -`fetch_wire_documents` uses one shared fetcher for the feed and article -requests. The result contains successful documents and recorded failures. - -```python -import asyncio - -from quantmind.preprocess import ( - PR_NEWSWIRE, - FetchPolicy, - WireFeedConfig, - fetch_wire_documents, -) - - -async def main() -> None: - result = await fetch_wire_documents( - WireFeedConfig( - provider=PR_NEWSWIRE, - feed_urls=( - "https://www.prnewswire.com/rss/news-releases-list.rss", - ), - fetch_policy=FetchPolicy(min_interval_seconds=0.25), - ) - ) - - print(f"documents={result.success_count}") - print(f"failures={result.failure_count}") - - if result.documents: - document = result.documents[0] - print(document.identity) - print(document.cleaned_markdown[:200]) - print(document.raw_feed_entry.content_hash) - if document.raw_article is not None: - print(document.raw_article.content_hash) - - for failure in result.failures: - print(failure.stage, failure.error_type, failure.source_url) - - -asyncio.run(main()) -``` - -A focused runnable version lives at -[`examples/preprocess/02_wire_ingestion.py`](../examples/preprocess/02_wire_ingestion.py). - -## Verification - -The normal test suite is deterministic and does not require public network -access: - -```bash -python -m pytest --no-cov tests/preprocess -bash scripts/verify.sh -``` - -The live smoke test is separate and performs real PR Newswire feed and article -requests: - -```bash -python scripts/smoke_wire.py -``` - -The smoke checklist verifies that the current public feed produces documents, -raw feed entries, raw articles, cleaned Markdown, unique identities, and no -recorded failures. - -## Current Non-Goals - -The current pipeline does not provide: - -- built-in GlobeNewswire or Business Wire adapters; -- persistent cursors, watermarks, or time-window pagination; -- scheduling or durable storage; -- a shared batch-operation base class, hooks, monitoring, or metrics; -- authenticated or private provider endpoints; -- automatic company or instrument resolution; -- downstream news-card generation. diff --git a/docs/README.md b/docs/README.md new file mode 100644 index 0000000..757ab4b --- /dev/null +++ b/docs/README.md @@ -0,0 +1,52 @@ +# QuantMind Component Catalog + +This is the discovery index for QuantMind's public operations, supported +sources, examples, design documents, and verification commands. Source-specific +acquisition mechanics remain internal unless they are intentionally documented +as a public preprocessing primitive. + +Public callable names follow the +[operation naming contract](design/en/operations.md). The runtime API serves +Python callers; coding-agent guidance lives in the repository development +harness. + +## Public Operations + +| Operation | Import | Input and config | Result | Example | Design or guide | +|---|---|---|---|---|---| +| Paper extraction | `quantmind.flows.paper_flow` | `PaperInput`, `PaperFlowCfg` | `Paper` | [README usage](../README.md#-usage-examples) | [Papers](papers.md) | +| News collection | `quantmind.flows.collect_news` | `NewsWindow`, `NewsCollectionCfg` | `NewsBatch` from `quantmind.preprocess` | [Collect news](../examples/flows/collect_news.py) | [News collection design](design/en/news.md) | +| Bounded fan-out | `quantmind.flows.batch_run` | Operation inputs and shared config | `BatchResult` | [README usage](../README.md#-usage-examples) | API docstrings | + +Import public inputs and configs from `quantmind.configs` and current public +operations from `quantmind.flows`. Import result contracts from the canonical +layer shown in the catalog. + +## Public-Network Sources + +| Source | Source selection | Operation | Live component gate | +|---|---|---|---| +| 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. + +## Verification + +Run the deterministic offline golden gate for every change: + +```bash +bash scripts/verify.sh +``` + +It covers formatting, linting, typing, import boundaries, unit tests, and +coverage, and must remain network-free. When a change affects a public-network +component, also run every applicable live gate listed above. + +## Adding a Public Operation or Source + +Use the `quantmind-dev` component workflow. A public operation is not complete +until its typed contract, package exports, offline tests, focused example, +design or guide, and catalog row agree. A public-network source additionally +needs mocked source tests plus a bounded live verifier and CI workflow. diff --git a/docs/design/en/news.md b/docs/design/en/news.md new file mode 100644 index 0000000..269a967 --- /dev/null +++ b/docs/design/en/news.md @@ -0,0 +1,252 @@ +# News Collection Design + +## Status and Scope + +This document defines the OSS contract for collecting public company news in +QuantMind. The MVP supports PR Newswire only. It is deliberately small: one +intent-oriented collection operation, one time-window input, deterministic +preprocessing, and explicit partial-failure reporting. + +Its public name follows the +[operation naming contract](operations.md): collection returns source-faithful +evidence and remains separate from semantic knowledge extraction. + +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. **Express intent, not acquisition mechanics.** Callers ask for news from a + source and time window. They do not select RSS, listing pages, pagination, + or article-body rules. +2. **Return honest observations.** QuantMind does not silently deduplicate + source rows. Repeated source observations remain repeated output records, + and may share the same stable identity. +3. **Make partial work inspectable.** Item failures are values in the returned + batch. Invalid inputs still raise before network work starts. +4. **Keep source policy internal.** PR Newswire discovery can change from + listing pages to another public mechanism without changing the collection + contract. +5. **Make the supported set explicit.** The MVP uses exhaustive source + dispatch. It does not expose a provider protocol or a provider registry. +6. **Separate collection from production policy.** Persistence, deduplication, + rule-based pruning, target schemas, and scheduling belong to the consuming + data pipeline. + +## Public API Contract + +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` uses timezone-aware timestamps and the half-open interval +`[start, end)`. Both regular collection and historical backfill use this same +call. There are no separate `poll_*`, `backfill_*`, or `fetch_wire_*` public +entry points. + +`NewsCollectionCfg` retains the repository's shared `BaseFlowCfg` fields so it +works with the common typed-operation and magic-input tooling. 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 +conservative and storage-efficient behavior. The article is still fetched, +hashed, parsed, and represented by metadata; only its byte payload is +discarded. The deterministic collector does not otherwise consume the shared +model, tracing, or SDK-run fields. + +### Collection records are not knowledge + +QuantMind keeps source evidence and semantic extraction as separate contracts: + +| Operation | Result | Canonical layer | +|-----------|--------|-----------------| +| `collect_news` | Source-faithful documents, artifacts, failures, and coverage | `quantmind.preprocess` | +| future `extract_news_knowledge` | Extracted financial events | `quantmind.knowledge.News` | + +`NewsDocument` is therefore not a `KnowledgeItem`. It carries HTTP evidence, +raw bytes, parsing output, and collection status that remain useful before any +LLM or business schema is chosen. `knowledge.News` is a compact semantic event +with entities, sentiment, materiality, provenance, and an embedding view. The +two may later be composed, but neither substitutes for the other. + +## Returned Data + +`collect_news` returns a `NewsBatch` with four concepts. The collection +contracts are publicly imported 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 normalized into + in-window observations, before article processing; +- `complete`: whether discovery proved coverage of the requested window. + +A `NewsDocument` contains the source name, stable identity, canonical URL, +title, publisher, publication time, cleaned Markdown, content hash, ticker +hints, and two evidence artifacts: + +- a small discovery artifact representing the public source row; +- an article artifact containing fetch metadata and a content hash. Its + `bytes` field is `None` unless `retain_raw_html=True`. + +`NewsArtifact` is the common evidence shape. It records 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 identity. The +consumer can use that identity for an idempotent upsert without losing what +QuantMind actually observed. + +## Collection Pipeline + +```text +NewsWindow + -> source dispatch + -> newest-to-oldest public discovery pages + -> observations inside [start, end) + -> linked article fetch + -> deterministic HTML-to-Markdown normalization + -> 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. This makes a past-day request independently replayable instead +of dependent on a previously saved cursor. + +PR Newswire exposes listing timestamps at minute precision. Scheduled callers +should therefore use minute-aligned window bounds, as the live E2E does. + +RSS remains a lower-level parser and a live component check. It is not a +high-level `NewsInput`, because feed selection is a source implementation +detail and a bounded feed snapshot cannot prove complete time-window coverage. + +The HTTP layer owns bounded retries, backoff, `Retry-After` handling, and +per-host rate limits. PR Newswire-specific URL construction and listing HTML +parsing stay in the PR Newswire source module. + +Supported source names form a closed set in `NewsWindow.source`. +`collect_news` dispatches them exhaustively, so widening the input schema +without implementing the corresponding collector fails static verification. + +## Failure and Completeness Semantics + +Configuration errors raise immediately. Examples include a naive timestamp, +an empty source, an unsupported source, or `end <= start`. + +After collection begins, recoverable failures are recorded and independent +items continue. Each `NewsFailure` identifies the source, processing stage, +URL, optional item identity, error category, and message. + +`complete=False` when any of the following is true: + +- a discovery page could not be fetched or parsed; +- discovery stopped before crossing the window start; + +Article failures remain explicit in `failures` but do not change discovery +completeness. A caller can therefore distinguish "the source window was fully +enumerated" from "every observed article was processed." It can safely persist +successful records while separately monitoring and replaying the failed +portion. An empty batch is not considered complete unless discovery has +positively crossed the requested start. + +## Responsibility Boundary + +QuantMind owns: + +- public-source discovery and article acquisition; +- deterministic normalization; +- stable identities and evidence hashes; +- honest batch counts, failure records, and completeness. + +The consuming production pipeline owns: + +- its schedule and GitHub Action invocation; +- durable storage, watermarks, and idempotent upserts; +- deduplication policy; +- rule-based company-news pruning; +- downstream schemas, enrichment, and database writes; +- shared batch hooks, metrics, and monitoring infrastructure. + +This boundary lets a separate ingestion job request the last day in one call, +apply its own rules, and write its own schema without coupling those policies +to this OSS library. + +## Verification Contract + +Verification has two intentionally separate layers. + +### Offline verification + +`bash scripts/verify.sh` runs deterministic unit tests, linting, typing, and +coverage. News tests use saved HTML/RSS fixtures and mocked HTTP responses. +They cover window validation, time boundaries, pagination, duplicate +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: + +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. + +GitHub Actions runs this check on every pull request, once daily, and on manual +dispatch. The ordinary offline verification workflow remains network-free so +local development and unit tests stay deterministic. + +## Non-Goals and Extension Rule + +The MVP does not include GlobeNewswire, Business Wire, authenticated feeds, +continuous cursors, storage, deduplication, business materiality scoring, or a +generic batch-operation base class. + +When a second source is implemented, compare its real behavior with PR +Newswire first. Extract a shared provider interface only for behavior the two +implementations genuinely share. The public +`collect_news(NewsWindow, *, cfg)` contract should remain unchanged. + +## Adding a Public News Source + +A coding agent adding a second source follows this closed 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 exhaustive type check must + remain green. +4. Add fixture-based success, boundary, duplicate-observation, + partial-failure, and completeness tests for the source. +5. Add a routing test proving 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 bounded live component verifier and GitHub workflow when + the integration depends on a public network endpoint. + +Only after two real collectors expose shared behavior should a common +`Protocol` be considered. A new source must never be added only to the input +Literal: static verification is expected to reject that incomplete change. diff --git a/docs/design/en/operations.md b/docs/design/en/operations.md new file mode 100644 index 0000000..c1866b0 --- /dev/null +++ b/docs/design/en/operations.md @@ -0,0 +1,62 @@ +# Public Operation Naming + +## Scope + +This document defines how public QuantMind callables communicate intent. It is +a naming contract, not a package-layout migration. Existing names may be +changed in focused compatibility work; they are not silently renamed here. + +The runtime library API is designed for Python callers. Repository guidance +for coding agents belongs to the development harness (`AGENTS.md`, skills, +docs, fixtures, and verification), not in the runtime API description. + +## Operation Stages + +Public names use a stage verb plus the domain or result: + +| Stage | Name pattern | Transformation | +|-------|--------------|----------------| +| Collection | `collect_` | External sources to source-faithful documents and evidence | +| Knowledge extraction | `extract__knowledge` | Documents to typed `quantmind.knowledge` values | +| Index construction | `build__index` | Documents or knowledge to a retrieval index | +| Analysis | `analyze_` | Domain inputs to an analytical result | +| Generic execution | `batch_run` and similar combinators | Apply an operation without changing its domain meaning | +| Composed recipe | `__pipeline` | Compose multiple named stages into a reusable pipeline | + +`flow` is not a domain verb. Do not add new public `*_flow` names merely +because a callable performs several steps. Use a precise operation name, or a +`*_pipeline` name only when the callable deliberately composes multiple public +stages as a reusable recipe. + +## Type Names + +- Input types describe caller intent, such as `NewsWindow` or `PaperInput`. +- Config types name the domain and stage, such as `NewsCollectionCfg` or a + future `PaperExtractionCfg`. +- Result types describe returned data, such as `NewsBatch`, `Paper`, or + `PageIndex`. +- Provider names stay out of public operation names unless provider-specific + behavior is itself the public contract. + +## Current API + +- `collect_news` is a collection operation and follows this contract. +- `batch_run` is a generic execution combinator, not a news or paper stage. +- `paper_flow` is an existing semantic-extraction API with legacy naming. It + is not the naming precedent for new operations; any rename belongs in a + separate compatibility change. + +The current `quantmind.flows` package remains the apex implementation namespace +for this release. Whether it should become `operations` or be split from a +future `pipelines` package is deliberately outside this document. + +## Review Checklist + +Before adding a public callable, state: + +1. Its operation stage. +2. Its input and result contracts. +3. Whether it is one operation or a genuine multi-stage pipeline. +4. Why its verb matches the observable result. + +If those answers are unclear, settle the contract before adding a public name. diff --git a/examples/flows/collect_news.py b/examples/flows/collect_news.py new file mode 100644 index 0000000..1b02a42 --- /dev/null +++ b/examples/flows/collect_news.py @@ -0,0 +1,28 @@ +"""Collect one replayable day of PR Newswire observations.""" + +import asyncio +from datetime import datetime, timedelta, timezone + +from quantmind.configs import NewsCollectionCfg, NewsWindow +from quantmind.flows import collect_news + + +async def main() -> None: + """Collect and summarize the preceding 24-hour window.""" + end = datetime.now(timezone.utc) + batch = await collect_news( + NewsWindow( + source="pr-newswire", + start=end - timedelta(days=1), + end=end, + ), + cfg=NewsCollectionCfg(retain_raw_html=False), + ) + print( + f"observed={batch.observed_count} documents={batch.success_count} " + f"failures={batch.failure_count} complete={batch.complete}" + ) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/preprocess/01_news_pr_wire.py b/examples/preprocess/01_news_pr_wire.py index ec211c9..ea66e33 100644 --- a/examples/preprocess/01_news_pr_wire.py +++ b/examples/preprocess/01_news_pr_wire.py @@ -19,6 +19,6 @@ candidate = preprocess_news_document(raw) -print(candidate.dedup_key) +print(candidate.identity) print(candidate.content_hash) print([hint.symbol for hint in candidate.ticker_hints]) diff --git a/examples/preprocess/02_wire_ingestion.py b/examples/preprocess/02_wire_ingestion.py deleted file mode 100644 index dd6058c..0000000 --- a/examples/preprocess/02_wire_ingestion.py +++ /dev/null @@ -1,32 +0,0 @@ -"""Fetch the public PR Newswire feed and print a compact summary.""" - -import asyncio - -from quantmind.preprocess import ( - PR_NEWSWIRE, - FetchPolicy, - WireFeedConfig, - fetch_wire_documents, -) - -_PR_NEWSWIRE_FEED = "https://www.prnewswire.com/rss/news-releases-list.rss" - - -async def main() -> None: - """Fetch PR Newswire and print a compact result summary.""" - result = await fetch_wire_documents( - WireFeedConfig( - provider=PR_NEWSWIRE, - feed_urls=(_PR_NEWSWIRE_FEED,), - fetch_policy=FetchPolicy(min_interval_seconds=0.25), - ) - ) - print(f"documents={result.success_count} failures={result.failure_count}") - if result.documents: - first = result.documents[0] - print(first.title) - print(first.identity) - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/quantmind/configs/__init__.py b/quantmind/configs/__init__.py index 271be37..2e9843d 100644 --- a/quantmind/configs/__init__.py +++ b/quantmind/configs/__init__.py @@ -1,7 +1,7 @@ -"""quantmind.configs — flow configuration + input types. +"""quantmind.configs — public operation configuration + input types. -Each flow has a `FlowCfg` (extends `BaseFlowCfg`) and a `Input` -discriminated-union type. All cfg / input classes live here so that: +Public operations use a ``BaseFlowCfg`` subclass and a typed input model or +discriminated union. All cfg / input classes live here so that: - YAML / CLI users see a single import surface, - JSON schemas can be exported uniformly (for IDE autocomplete), @@ -10,7 +10,7 @@ from quantmind.configs.base import BaseFlowCfg, BaseInput from quantmind.configs.earnings import EarningsFlowCfg, EarningsInput -from quantmind.configs.news import NewsFlowCfg, NewsInput +from quantmind.configs.news import NewsCollectionCfg, NewsWindow from quantmind.configs.paper import PaperFlowCfg, PaperInput __all__ = [ @@ -18,8 +18,8 @@ "BaseInput", "EarningsFlowCfg", "EarningsInput", - "NewsFlowCfg", - "NewsInput", + "NewsCollectionCfg", + "NewsWindow", "PaperFlowCfg", "PaperInput", ] diff --git a/quantmind/configs/news.py b/quantmind/configs/news.py index 3421c7d..ea71cc7 100644 --- a/quantmind/configs/news.py +++ b/quantmind/configs/news.py @@ -1,41 +1,37 @@ -"""News-flow configuration + input discriminated union.""" +"""Typed input and configuration for deterministic news collection.""" -from typing import Annotated, Literal, Union +from datetime import timezone +from typing import Literal -from pydantic import Field +from pydantic import AwareDatetime, field_validator, model_validator +from typing_extensions import Self from quantmind.configs.base import BaseFlowCfg, BaseInput -class RssFeed(BaseInput): - """RSS/Atom feed URL to be polled for items.""" +class NewsWindow(BaseInput): + """A replayable source window using the half-open interval [start, end).""" - type: Literal["rss"] = "rss" - url: str + type: Literal["window"] = "window" + source: Literal["pr-newswire"] + start: AwareDatetime + end: AwareDatetime + @field_validator("start", "end") + @classmethod + def normalize_to_utc(cls, value: AwareDatetime) -> AwareDatetime: + """Normalize aware timestamps so collectors receive one timezone.""" + return value.astimezone(timezone.utc) -class HttpUrl(BaseInput): - """Single news article URL.""" + @model_validator(mode="after") + def validate_order(self) -> Self: + """Reject empty or reversed collection windows.""" + if self.end <= self.start: + raise ValueError("NewsWindow requires end to be after start") + return self - type: Literal["http"] = "http" - url: str +class NewsCollectionCfg(BaseFlowCfg): + """Collection behavior that changes the returned evidence.""" -class Headline(BaseInput): - """Inline headline text (no body fetching).""" - - type: Literal["headline"] = "headline" - text: str - - -NewsInput = Annotated[ - Union[RssFeed, HttpUrl, Headline], - Field(discriminator="type"), -] - - -class NewsFlowCfg(BaseFlowCfg): - """Knobs specific to news_flow.""" - - materiality_threshold: Literal["low", "medium", "high"] = "medium" - entities_hint: list[str] = Field(default_factory=list) + retain_raw_html: bool = False diff --git a/quantmind/flows/__init__.py b/quantmind/flows/__init__.py index 78dd9bf..6290275 100644 --- a/quantmind/flows/__init__.py +++ b/quantmind/flows/__init__.py @@ -1,8 +1,8 @@ """Apex layer — composes configs / knowledge / preprocess on the SDK. -Each flow function (``paper_flow``, future ``news_flow`` / ``earnings_flow``) -takes a typed input and a ``FlowCfg`` and returns a knowledge item. -Cross-flow utilities live alongside: +Semantic flows such as ``paper_flow`` return ``quantmind.knowledge`` values. +Deterministic operations such as ``collect_news`` return source-faithful +``quantmind.preprocess`` values. Cross-flow utilities live alongside: - ``batch_run`` runs any flow over a list of inputs with bounded concurrency and aggregated results. @@ -12,11 +12,13 @@ """ from quantmind.flows.batch import BatchResult, batch_run +from quantmind.flows.news import collect_news from quantmind.flows.paper import UnsupportedContentTypeError, paper_flow __all__ = [ "BatchResult", "UnsupportedContentTypeError", "batch_run", + "collect_news", "paper_flow", ] diff --git a/quantmind/flows/news.py b/quantmind/flows/news.py new file mode 100644 index 0000000..d8af079 --- /dev/null +++ b/quantmind/flows/news.py @@ -0,0 +1,37 @@ +"""Intent-oriented collection operation for public company news.""" + +from typing_extensions import assert_never + +from quantmind.configs import NewsCollectionCfg, NewsWindow +from quantmind.preprocess import NewsBatch +from quantmind.preprocess.pr_newswire import _collect_pr_newswire + +__all__ = ["collect_news"] + + +async def collect_news( + input: NewsWindow, + *, + cfg: NewsCollectionCfg | None = None, +) -> NewsBatch: + """Collect one replayable window without exposing source mechanics. + + Args: + input: Source and half-open time window to collect. + cfg: Options that change the returned collection evidence. + + Returns: + Successfully collected documents, recoverable failures, and whether + source discovery proved complete coverage of the requested window. + Article failures do not make discovery incomplete. + """ + cfg = cfg or NewsCollectionCfg() + source = input.source + if source == "pr-newswire": + return await _collect_pr_newswire( + start=input.start, + end=input.end, + retain_raw_html=cfg.retain_raw_html, + ) + + assert_never(source) diff --git a/quantmind/knowledge/news.py b/quantmind/knowledge/news.py index 6604daa..2bf2de9 100644 --- a/quantmind/knowledge/news.py +++ b/quantmind/knowledge/news.py @@ -1,4 +1,4 @@ -"""News knowledge schema (output of news_flow).""" +"""Semantic news-event schema for agent and LLM extraction.""" from datetime import datetime from typing import Literal @@ -9,7 +9,7 @@ class News(FlattenKnowledge): - """A single news event extraction.""" + """A single extracted news event, distinct from collection evidence.""" item_type: Literal["news"] = "news" diff --git a/quantmind/magic.py b/quantmind/magic.py index 71e522a..819b96c 100644 --- a/quantmind/magic.py +++ b/quantmind/magic.py @@ -17,7 +17,15 @@ import json import types from collections.abc import Awaitable, Callable -from typing import Any, Generic, TypeVar, Union, get_args, get_origin +from typing import ( + Any, + Generic, + TypeVar, + Union, + get_args, + get_origin, + get_type_hints, +) from agents import Agent, Runner from pydantic import BaseModel @@ -120,10 +128,10 @@ def _introspect_flow_signature( ) -> tuple[Any, type[BaseFlowCfg]]: """Return ``(input_annotation, cfg_type)`` for a flow function. - ``input_annotation`` is returned as-is — it may be a discriminated- - union alias such as ``Annotated[Union[...], Field(discriminator=...)]``. - Pydantic accepts both plain ``BaseModel`` subclasses and discriminated - aliases as generic parameters. + Annotations are resolved before inspection, including postponed string + annotations. ``input_annotation`` may be a discriminated-union alias such + as ``Annotated[Union[...], Field(discriminator=...)]``; its extra metadata + is preserved for Pydantic. ``cfg_type`` strips an outer ``T | None`` so the resolver instantiates the concrete cfg subclass. The result must be a ``BaseFlowCfg`` @@ -138,8 +146,9 @@ def _introspect_flow_signature( raise TypeError( f"Flow {flow_fn.__name__!r} must accept a `cfg` keyword parameter" ) - input_anno = sig.parameters["input"].annotation - cfg_anno = sig.parameters["cfg"].annotation + annotations = get_type_hints(flow_fn, include_extras=True) + input_anno = annotations.get("input", sig.parameters["input"].annotation) + cfg_anno = annotations.get("cfg", sig.parameters["cfg"].annotation) cfg_type = _strip_optional(cfg_anno) if not (isinstance(cfg_type, type) and issubclass(cfg_type, BaseFlowCfg)): raise TypeError( diff --git a/quantmind/preprocess/__init__.py b/quantmind/preprocess/__init__.py index f62878b..97ca7ed 100644 --- a/quantmind/preprocess/__init__.py +++ b/quantmind/preprocess/__init__.py @@ -1,5 +1,13 @@ """Preprocess layer for fetching, formatting, and source normalization.""" +from quantmind.preprocess._news_types import ( + NewsArtifact, + NewsBatch, + NewsDocument, + NewsFailure, + NewsFailureStage, + NewsTickerHint, +) from quantmind.preprocess.clean import ( collapse_whitespace, dedupe_lines, @@ -30,10 +38,9 @@ from quantmind.preprocess.news import ( BodySource, NewsCandidate, - NewsTickerHint, RawNewsDocument, - build_news_dedup_key, - build_sec_news_dedup_key, + build_news_identity, + build_sec_news_identity, canonicalize_source_url, extract_exchange_ticker_hints, feed_item_to_news_document, @@ -51,18 +58,6 @@ parse_news_datetime, to_utc, ) -from quantmind.preprocess.wire import ( - PR_NEWSWIRE, - RawWireArtifact, - WireDocument, - WireFeedConfig, - WireFetchFailure, - WireFetchResult, - WireItemMapping, - WireProvider, - build_wire_identity, - fetch_wire_documents, -) __all__ = [ "ArxivIdParseError", @@ -73,23 +68,19 @@ "FetchPolicy", "Fetched", "HttpFetcher", + "NewsArtifact", + "NewsBatch", "NewsCandidate", + "NewsDocument", + "NewsFailure", + "NewsFailureStage", "NewsTickerHint", - "PR_NEWSWIRE", "PdfParseError", "RawFeed", "RawNewsDocument", "RawPaper", - "RawWireArtifact", - "WireDocument", - "WireFeedConfig", - "WireFetchFailure", - "WireFetchResult", - "WireItemMapping", - "WireProvider", - "build_news_dedup_key", - "build_sec_news_dedup_key", - "build_wire_identity", + "build_news_identity", + "build_sec_news_identity", "business_days_between", "canonicalize_source_url", "collapse_whitespace", @@ -100,7 +91,6 @@ "fetch_news_document", "fetch_rss_feed", "fetch_url", - "fetch_wire_documents", "html_to_markdown", "news_content_hash", "news_document_from_fetched", diff --git a/quantmind/preprocess/_news_types.py b/quantmind/preprocess/_news_types.py new file mode 100644 index 0000000..d41f6c8 --- /dev/null +++ b/quantmind/preprocess/_news_types.py @@ -0,0 +1,91 @@ +"""Source-faithful news collection values, exported by ``preprocess``. + +These records carry acquisition evidence and status. They are intentionally +separate from semantic ``quantmind.knowledge.News`` values. +""" + +from dataclasses import dataclass, field +from datetime import datetime +from typing import Literal + +NewsFailureStage = Literal[ + "discovery_fetch", + "discovery_parse", + "article_fetch", + "article_parse", +] + + +@dataclass(frozen=True, slots=True) +class NewsTickerHint: + """Ticker hint extracted before instrument resolution.""" + + symbol: str + exchange: str | None = None + source: str = "exchange_code" + confidence: float = 1.0 + raw: str | None = None + + +@dataclass(frozen=True, slots=True) +class NewsArtifact: + """Raw evidence and fetch metadata retained for replay or auditing.""" + + bytes: bytes | None + content_hash: str + content_type: str + source_url: str | None + resolved_url: str | None + status_code: int | None + headers: dict[str, str] = field(default_factory=dict) + fetched_at: datetime | None = None + + +@dataclass(frozen=True, slots=True) +class NewsDocument: + """One source observation with cleaned text and its raw evidence.""" + + source: str + identity: str + cleaned_markdown: str + content_hash: str + discovery_artifact: NewsArtifact + article_artifact: NewsArtifact + payload_id: str | None = None + canonical_url: str | None = None + title: str | None = None + publisher: str | None = None + published_at: datetime | None = None + ticker_hints: tuple[NewsTickerHint, ...] = () + + +@dataclass(frozen=True, slots=True) +class NewsFailure: + """Lightweight record for one recoverable collection failure.""" + + source: str + stage: NewsFailureStage + source_url: str + item_id: str | None + error_type: str + message: str + + +@dataclass(frozen=True, slots=True) +class NewsBatch: + """Observed documents, failures, and discovery-coverage status.""" + + documents: tuple[NewsDocument, ...] = () + failures: tuple[NewsFailure, ...] = () + observed_count: int = 0 + complete: bool = False + + @property + def success_count(self) -> int: + """Number of successfully collected observations.""" + return len(self.documents) + + @property + def failure_count(self) -> int: + """Number of recorded recoverable failures.""" + return len(self.failures) diff --git a/quantmind/preprocess/news.py b/quantmind/preprocess/news.py index 352df6d..ac36346 100644 --- a/quantmind/preprocess/news.py +++ b/quantmind/preprocess/news.py @@ -7,6 +7,7 @@ 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, @@ -68,17 +69,6 @@ } -@dataclass(frozen=True, slots=True) -class NewsTickerHint: - """Ticker hint extracted before instrument resolution.""" - - symbol: str - exchange: str | None = None - source: str = "exchange_code" - confidence: float = 1.0 - raw: str | None = None - - @dataclass(frozen=True, slots=True) class RawNewsDocument: """Raw news document ready for source-agnostic preprocessing.""" @@ -100,7 +90,7 @@ class NewsCandidate: body_text: str content_hash: str source_type: NewsSourceType - dedup_key: str + identity: str source_url: str | None = None title: str | None = None publisher: str | None = None @@ -319,7 +309,7 @@ def preprocess_news_document(raw: RawNewsDocument) -> NewsCandidate: body_text=body_text, content_hash=news_content_hash(body_text), source_type=raw.source_type, - dedup_key=build_news_dedup_key( + identity=build_news_identity( source_type=raw.source_type, source_url=source_url, payload_id=raw.payload_id, @@ -345,13 +335,13 @@ def news_content_hash(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest() -def build_news_dedup_key( +def build_news_identity( *, source_type: NewsSourceType, source_url: str | None = None, payload_id: str | None = None, ) -> str: - """Build a deterministic source-document dedup key. + """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 @@ -361,27 +351,27 @@ def build_news_dedup_key( if not identity and source_url: identity = canonicalize_source_url(source_url) if not identity: - raise ValueError("news dedup key requires payload_id or source_url") + 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_dedup_key( +def build_sec_news_identity( *, accession_number: str, section_key: str, ) -> str: - """Build a stable 8-K EX-99.x news dedup key.""" + """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 dedup key requires accession and 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 dedup identity.""" + """Normalise source URLs for stable source identity.""" parsed = urlsplit(url.strip()) query = [ (key, value) diff --git a/quantmind/preprocess/pr_newswire.py b/quantmind/preprocess/pr_newswire.py new file mode 100644 index 0000000..8f0e122 --- /dev/null +++ b/quantmind/preprocess/pr_newswire.py @@ -0,0 +1,733 @@ +"""Internal PR Newswire source implementation for ``collect_news``.""" + +import asyncio +import hashlib +import re +import time +from dataclasses import dataclass +from datetime import date, datetime, timedelta, timezone +from html import unescape +from html.parser import HTMLParser +from urllib.parse import urlencode, urljoin +from zoneinfo import ZoneInfo + +import httpx + +from quantmind.preprocess._news_types import ( + NewsArtifact, + NewsBatch, + NewsDocument, + NewsFailure, + NewsFailureStage, +) +from quantmind.preprocess.fetch._types import Fetched +from quantmind.preprocess.fetch.http import ( + FetchAttemptsExhausted, + FetchPolicy, + HttpFetcher, +) +from quantmind.preprocess.news import ( + canonicalize_source_url, + news_document_from_fetched, + preprocess_news_document, +) + +_SOURCE = "pr-newswire" +_PUBLISHER = "PR Newswire" +_BASE_URL = "https://www.prnewswire.com" +_LISTING_URL = f"{_BASE_URL}/news-releases/news-releases-list/" +_EASTERN = ZoneInfo("America/New_York") +_PAGE_SIZE = 100 +_MAX_PAGES = 20 +_CACHE_BUST_ATTEMPTS = 8 +_ARTICLE_MAX_BYTES = 10_000_000 +_LISTING_MAX_BYTES = 5_000_000 +_DEFAULT_FETCH_POLICY = FetchPolicy( + max_attempts=3, + backoff_base_seconds=0.5, + backoff_max_seconds=5.0, + jitter_seconds=0.1, + max_concurrency_per_host=2, + min_interval_seconds=0.1, +) + +_SHORT_TIME_RE = re.compile( + r"^(?P\d{1,2}):(?P\d{2})\s+E(?:S|D)?T$", + re.IGNORECASE, +) +_FULL_TIME_RE = re.compile( + r"^(?P[A-Za-z]{3})\s+(?P\d{1,2}),\s+" + r"(?P\d{4}),\s+(?P\d{1,2}):" + r"(?P\d{2})\s+E(?:S|D)?T$", + re.IGNORECASE, +) +_MONTHS = { + month: number + for number, month in enumerate( + ( + "jan", + "feb", + "mar", + "apr", + "may", + "jun", + "jul", + "aug", + "sep", + "oct", + "nov", + "dec", + ), + start=1, + ) +} +_PAYLOAD_ID_RE = re.compile(r"-(?P\d+)\.html$") + + +@dataclass(frozen=True, slots=True) +class PRNewswireObservation: + """One public listing row inside the requested time window.""" + + identity: str + payload_id: str | None + canonical_url: str + title: str + published_at: datetime + discovery_artifact: NewsArtifact + + +@dataclass(frozen=True, slots=True) +class PRNewswireDiscovery: + """Listing observations and evidence that window discovery completed.""" + + observations: tuple[PRNewswireObservation, ...] = () + failures: tuple[NewsFailure, ...] = () + page_count: int = 0 + complete: bool = False + + @property + def observed_count(self) -> int: + """Number of listing rows retained inside the requested window.""" + return len(self.observations) + + +@dataclass(frozen=True, slots=True) +class _ParsedListingRow: + href: str | None + title: str + timestamp: str + raw_html: bytes + + +@dataclass(frozen=True, slots=True) +class _ParsedListingPage: + page_date: date + rows: tuple[_ParsedListingRow, ...] + + +class _ListingParser(HTMLParser): + """Extract only PR Newswire listing cards with no optional dependency.""" + + def __init__(self) -> None: + super().__init__(convert_charrefs=False) + self.page_date_text: str | None = None + self.rows: list[_ParsedListingRow] = [] + self._row_div_depth = 0 + self._raw_parts: list[str] = [] + self._href: str | None = None + self._title_parts: list[str] = [] + self._timestamp_parts: list[str] = [] + self._in_heading = False + self._in_timestamp = False + + def handle_starttag( + self, + tag: str, + attrs: list[tuple[str, str | None]], + ) -> None: + attributes = dict(attrs) + classes = set((attributes.get("class") or "").split()) + if ( + not self._row_div_depth + and tag == "div" + and {"row", "newsCards"}.issubset(classes) + ): + self._start_row() + + if not self._row_div_depth: + if tag == "input" and attributes.get("id") == "date": + self.page_date_text = attributes.get("value") + return + + start_text = self.get_starttag_text() or f"<{tag}>" + self._raw_parts.append(start_text) + if tag == "div" and len(self._raw_parts) > 1: + self._row_div_depth += 1 + if tag == "a" and "newsreleaseconsolidatelink" in classes: + self._href = self._href or attributes.get("href") + if tag == "h3": + self._in_heading = True + if tag == "small" and self._in_heading: + self._in_timestamp = True + + def handle_startendtag( + self, + tag: str, + attrs: list[tuple[str, str | None]], + ) -> None: + attributes = dict(attrs) + if not self._row_div_depth: + if tag == "input" and attributes.get("id") == "date": + self.page_date_text = attributes.get("value") + return + self._raw_parts.append(self.get_starttag_text() or f"<{tag}/>") + + def handle_endtag(self, tag: str) -> None: + if not self._row_div_depth: + return + self._raw_parts.append(f"") + if tag == "small": + self._in_timestamp = False + if tag == "h3": + self._in_heading = False + self._in_timestamp = False + if tag == "div": + self._row_div_depth -= 1 + if not self._row_div_depth: + self._finish_row() + + def handle_data(self, data: str) -> None: + if not self._row_div_depth: + return + self._raw_parts.append(data) + self._append_heading_text(data) + + def handle_entityref(self, name: str) -> None: + self._handle_reference(f"&{name};") + + def handle_charref(self, name: str) -> None: + self._handle_reference(f"&#{name};") + + def handle_comment(self, data: str) -> None: + if self._row_div_depth: + self._raw_parts.append(f"") + + def _start_row(self) -> None: + self._row_div_depth = 1 + self._raw_parts = [] + self._href = None + self._title_parts = [] + self._timestamp_parts = [] + self._in_heading = False + self._in_timestamp = False + + def _finish_row(self) -> None: + self.rows.append( + _ParsedListingRow( + href=self._href, + title=_normalize_space(unescape("".join(self._title_parts))), + timestamp=_normalize_space( + unescape("".join(self._timestamp_parts)) + ), + raw_html="".join(self._raw_parts).encode("utf-8"), + ) + ) + self._raw_parts = [] + + def _handle_reference(self, reference: str) -> None: + if not self._row_div_depth: + return + self._raw_parts.append(reference) + self._append_heading_text(reference) + + def _append_heading_text(self, text: str) -> None: + if not self._in_heading: + return + if self._in_timestamp: + self._timestamp_parts.append(text) + else: + self._title_parts.append(text) + + +async def _discover_pr_newswire( + *, + start: datetime, + end: datetime, +) -> PRNewswireDiscovery: + """Discover every public PR Newswire listing row in ``[start, end)``.""" + start, end = _validate_window(start, end) + async with HttpFetcher(policy=_DEFAULT_FETCH_POLICY) as fetcher: + return await _discover_with_fetcher( + start=start, + end=end, + fetcher=fetcher, + ) + + +async def _collect_pr_newswire( + *, + start: datetime, + end: datetime, + retain_raw_html: bool, +) -> NewsBatch: + """Discover and collect PR Newswire articles for ``[start, end)``.""" + start, end = _validate_window(start, end) + async with HttpFetcher(policy=_DEFAULT_FETCH_POLICY) as fetcher: + discovery = await _discover_with_fetcher( + start=start, + end=end, + fetcher=fetcher, + ) + outcomes = await asyncio.gather( + *( + _collect_observation( + observation, + fetcher=fetcher, + retain_raw_html=retain_raw_html, + ) + for observation in discovery.observations + ) + ) + + documents = tuple( + outcome for outcome in outcomes if isinstance(outcome, NewsDocument) + ) + article_failures = tuple( + outcome for outcome in outcomes if isinstance(outcome, NewsFailure) + ) + return NewsBatch( + documents=documents, + failures=discovery.failures + article_failures, + observed_count=discovery.observed_count, + complete=discovery.complete, + ) + + +async def _discover_with_fetcher( + *, + start: datetime, + end: datetime, + fetcher: HttpFetcher, +) -> PRNewswireDiscovery: + observations: list[PRNewswireObservation] = [] + failures: list[NewsFailure] = [] + page_count = 0 + crossed_start = False + anchor = _ceil_to_hour(end).astimezone(_EASTERN) + prior_page: tuple[tuple[str | None, str], ...] | None = None + last_published_at: datetime | None = None + + for page_number in range(1, _MAX_PAGES + 1): + try: + fetched = await _fetch_listing_page( + fetcher, + anchor=anchor, + page_number=page_number, + ) + except Exception as exc: + failures.append( + _failure( + stage="discovery_fetch", + source_url=_listing_url( + anchor=anchor, + page_number=page_number, + ), + item_id=None, + error=exc, + ) + ) + break + + page_count += 1 + try: + parsed = _parse_listing_page(fetched, fallback_date=anchor.date()) + except Exception as exc: + failures.append( + _failure( + stage="discovery_parse", + source_url=( + fetched.resolved_url + or fetched.source_url + or _LISTING_URL + ), + item_id=None, + error=exc, + ) + ) + break + + fingerprint = tuple((row.href, row.timestamp) for row in parsed.rows) + if not fingerprint: + failures.append( + _failure( + stage="discovery_parse", + source_url=( + fetched.resolved_url + or fetched.source_url + or _LISTING_URL + ), + item_id=None, + error=ValueError("PR Newswire listing contained no rows"), + ) + ) + break + if fingerprint == prior_page: + failures.append( + _failure( + stage="discovery_parse", + source_url=( + fetched.resolved_url + or fetched.source_url + or _LISTING_URL + ), + item_id=None, + error=ValueError("PR Newswire pagination did not advance"), + ) + ) + break + prior_page = fingerprint + + for row in parsed.rows: + try: + published_at = _parse_listing_timestamp( + row.timestamp, + page_date=parsed.page_date, + previous=last_published_at, + ) + last_published_at = published_at + observation = _observation_from_row( + row, + published_at=published_at, + fetched_page=fetched, + ) + except Exception as exc: + failures.append( + _failure( + stage="discovery_parse", + source_url=( + fetched.resolved_url + or fetched.source_url + or _LISTING_URL + ), + item_id=row.href, + error=exc, + ) + ) + continue + + if start <= published_at < end: + observations.append(observation) + if published_at < start: + crossed_start = True + + if crossed_start: + break + else: + failures.append( + _failure( + stage="discovery_parse", + source_url=_LISTING_URL, + item_id=None, + error=ValueError( + f"PR Newswire discovery exceeded {_MAX_PAGES} pages" + ), + ) + ) + + return PRNewswireDiscovery( + observations=tuple(observations), + failures=tuple(failures), + page_count=page_count, + complete=crossed_start and not failures, + ) + + +async def _fetch_listing_page( + fetcher: HttpFetcher, + *, + anchor: datetime, + page_number: int, +) -> Fetched: + last_error: Exception | None = None + for _ in range(_CACHE_BUST_ATTEMPTS): + url = _listing_url( + anchor=anchor, + page_number=page_number, + cache_bust=str(time.time_ns()), + ) + try: + return await fetcher.fetch_url(url, max_bytes=_LISTING_MAX_BYTES) + except httpx.HTTPStatusError as exc: + last_error = exc + if exc.response.status_code != 404: + raise + assert last_error is not None + raise last_error + + +async def _collect_observation( + observation: PRNewswireObservation, + *, + fetcher: HttpFetcher, + retain_raw_html: bool, +) -> NewsDocument | NewsFailure: + try: + fetched = await fetcher.fetch_url( + observation.canonical_url, + max_bytes=_ARTICLE_MAX_BYTES, + ) + except Exception as exc: + return _failure( + stage="article_fetch", + source_url=observation.canonical_url, + item_id=observation.payload_id, + error=exc, + ) + + try: + raw = await news_document_from_fetched( + fetched, + title=observation.title, + publisher=_PUBLISHER, + published_at=observation.published_at, + payload_id=observation.payload_id, + ) + candidate = preprocess_news_document(raw) + except Exception as exc: + return _failure( + stage="article_parse", + source_url=observation.canonical_url, + item_id=observation.payload_id, + error=exc, + ) + + return NewsDocument( + source=_SOURCE, + identity=observation.identity, + cleaned_markdown=candidate.body_text, + content_hash=candidate.content_hash, + discovery_artifact=observation.discovery_artifact, + article_artifact=_artifact_from_fetched( + fetched, + retain_bytes=retain_raw_html, + ), + payload_id=observation.payload_id, + canonical_url=observation.canonical_url, + title=candidate.title, + publisher=candidate.publisher, + published_at=candidate.published_at, + ticker_hints=candidate.ticker_hints, + ) + + +def _parse_listing_page( + fetched: Fetched, + *, + fallback_date: date, +) -> _ParsedListingPage: + parser = _ListingParser() + try: + parser.feed(fetched.bytes.decode("utf-8")) + parser.close() + except (UnicodeDecodeError, ValueError) as exc: + raise ValueError(f"invalid PR Newswire listing HTML: {exc}") from exc + + page_date = fallback_date + if parser.page_date_text: + try: + page_date = datetime.strptime( + parser.page_date_text, + "%m/%d/%Y", + ).date() + except ValueError as exc: + raise ValueError( + f"invalid PR Newswire page date: {parser.page_date_text!r}" + ) from exc + return _ParsedListingPage(page_date=page_date, rows=tuple(parser.rows)) + + +def _observation_from_row( + row: _ParsedListingRow, + *, + published_at: datetime, + fetched_page: Fetched, +) -> PRNewswireObservation: + if not row.href: + raise ValueError("PR Newswire listing row has no release URL") + if not row.title: + raise ValueError("PR Newswire listing row has no title") + + canonical_url = canonicalize_source_url(urljoin(_BASE_URL, row.href)) + payload_match = _PAYLOAD_ID_RE.search(canonical_url) + payload_id = payload_match.group("id") if payload_match else None + identity = _build_identity( + payload_id=payload_id, + canonical_url=canonical_url, + ) + return PRNewswireObservation( + identity=identity, + payload_id=payload_id, + canonical_url=canonical_url, + title=row.title, + published_at=published_at, + discovery_artifact=NewsArtifact( + bytes=row.raw_html, + content_hash=hashlib.sha256(row.raw_html).hexdigest(), + content_type="text/html", + source_url=fetched_page.source_url, + resolved_url=fetched_page.resolved_url, + status_code=fetched_page.status_code, + headers=dict(fetched_page.headers), + fetched_at=fetched_page.fetched_at, + ), + ) + + +def _parse_listing_timestamp( + value: str, + *, + page_date: date, + previous: datetime | None, +) -> datetime: + previous_local = previous.astimezone(_EASTERN) if previous else None + short_match = _SHORT_TIME_RE.fullmatch(value) + if short_match: + local_date = previous_local.date() if previous_local else page_date + hour = int(short_match.group("hour")) + minute = int(short_match.group("minute")) + else: + full_match = _FULL_TIME_RE.fullmatch(value) + if not full_match: + raise ValueError(f"invalid PR Newswire timestamp: {value!r}") + month_name = full_match.group("month").lower() + try: + month = _MONTHS[month_name] + except KeyError as exc: + raise ValueError( + f"invalid PR Newswire month: {month_name!r}" + ) from exc + local_date = date( + int(full_match.group("year")), + month, + int(full_match.group("day")), + ) + hour = int(full_match.group("hour")) + minute = int(full_match.group("minute")) + + local = datetime.combine( + local_date, + datetime.min.time(), + tzinfo=_EASTERN, + ).replace(hour=hour, minute=minute) + if ( + short_match + and previous_local is not None + and (hour, minute) > (previous_local.hour, previous_local.minute) + ): + local -= timedelta(days=1) + return local.astimezone(timezone.utc) + + +def _listing_url( + *, + anchor: datetime, + page_number: int, + cache_bust: str | None = None, +) -> str: + params = { + "page": str(page_number), + "pagesize": str(_PAGE_SIZE), + "month": str(anchor.month), + "day": str(anchor.day), + "year": str(anchor.year), + "hour": f"{anchor.hour:02d}", + } + if cache_bust is not None: + params["_"] = cache_bust + return f"{_LISTING_URL}?{urlencode(params)}" + + +def _ceil_to_hour(value: datetime) -> datetime: + value = value.astimezone(timezone.utc) + if value.minute or value.second or value.microsecond: + return value.replace(minute=0, second=0, microsecond=0) + timedelta( + hours=1 + ) + return value + + +def _validate_window( + start: datetime, + end: datetime, +) -> tuple[datetime, datetime]: + if start.tzinfo is None or start.utcoffset() is None: + raise ValueError("PR Newswire discovery requires an aware start") + if end.tzinfo is None or end.utcoffset() is None: + raise ValueError("PR Newswire discovery requires an aware end") + start = start.astimezone(timezone.utc) + end = end.astimezone(timezone.utc) + if end <= start: + raise ValueError("PR Newswire discovery requires end after start") + return start, end + + +def _artifact_from_fetched( + fetched: Fetched, + *, + retain_bytes: bool, +) -> NewsArtifact: + return NewsArtifact( + bytes=fetched.bytes if retain_bytes else None, + content_hash=hashlib.sha256(fetched.bytes).hexdigest(), + content_type=fetched.content_type, + source_url=fetched.source_url, + resolved_url=fetched.resolved_url, + status_code=fetched.status_code, + headers=dict(fetched.headers), + fetched_at=fetched.fetched_at, + ) + + +def _build_identity( + *, + payload_id: str | None, + canonical_url: str, +) -> str: + raw = "\x1f".join((_SOURCE, payload_id or canonical_url)) + digest = hashlib.sha256(raw.encode("utf-8")).hexdigest() + return f"news:{_SOURCE}:{digest}" + + +def _failure( + *, + stage: NewsFailureStage, + source_url: str, + item_id: str | None, + error: Exception, +) -> NewsFailure: + return NewsFailure( + source=_SOURCE, + stage=stage, + source_url=source_url, + item_id=item_id, + error_type=_error_type(error), + message=str(error), + ) + + +def _error_type(error: Exception) -> str: + if isinstance(error, FetchAttemptsExhausted): + return "retry_exhausted" + if isinstance(error, httpx.TimeoutException): + return "timeout" + if isinstance(error, httpx.TransportError): + return "network" + if isinstance(error, httpx.HTTPStatusError): + return "http_status" + if isinstance(error, ValueError): + return "invalid_content" + return "unexpected" + + +def _normalize_space(value: str) -> str: + return " ".join(value.split()) diff --git a/quantmind/preprocess/wire.py b/quantmind/preprocess/wire.py deleted file mode 100644 index 1a32d50..0000000 --- a/quantmind/preprocess/wire.py +++ /dev/null @@ -1,431 +0,0 @@ -"""Provider-pluggable ingestion for public press-release wire feeds.""" - -import hashlib -from dataclasses import dataclass, field -from datetime import datetime -from typing import Literal, Protocol - -import httpx - -from quantmind.preprocess.fetch._types import Fetched -from quantmind.preprocess.fetch.http import ( - FetchAttemptsExhausted, - FetchPolicy, - HttpFetcher, -) -from quantmind.preprocess.fetch.rss import FeedItem, parse_feed -from quantmind.preprocess.news import ( - BodySource, - NewsTickerHint, - canonicalize_source_url, - feed_item_to_news_document, - news_document_from_fetched, - preprocess_news_document, -) - -WireFailureStage = Literal[ - "feed_fetch", - "feed_parse", - "article_fetch", - "article_parse", - "document_build", -] - - -@dataclass(frozen=True, slots=True) -class WireItemMapping: - """Provider-normalized identity and metadata for one feed item.""" - - payload_id: str | None - canonical_url: str | None - title: str - published_at: datetime | None - - -class WireProvider(Protocol): - """Adapter contract for provider-specific feed item behavior.""" - - @property - def name(self) -> str: - """Stable provider identifier.""" - ... - - @property - def publisher(self) -> str: - """Human-readable publisher name.""" - ... - - @property - def body_source(self) -> BodySource: - """Whether cleaned body text comes from feed or article HTML.""" - ... - - def map_item(self, item: FeedItem) -> WireItemMapping: - """Map one provider feed item into common identity fields.""" - ... - - -@dataclass(frozen=True, slots=True) -class _StandardWireProvider: - name: str - publisher: str - body_source: BodySource - - def map_item(self, item: FeedItem) -> WireItemMapping: - url = canonicalize_source_url(item.url) if item.url else None - return WireItemMapping( - payload_id=(item.id or "").strip() or None, - canonical_url=url, - title=item.title, - published_at=item.published_at, - ) - - -PR_NEWSWIRE: WireProvider = _StandardWireProvider( - name="pr-newswire", - publisher="PR Newswire", - body_source="article", -) - - -@dataclass(frozen=True, slots=True) -class WireFeedConfig: - """Public feed URLs, provider adapter, and shared HTTP policy.""" - - provider: WireProvider - feed_urls: tuple[str, ...] - fetch_policy: FetchPolicy = field(default_factory=FetchPolicy) - - def __post_init__(self) -> None: - """Reject incomplete configs before network calls begin.""" - if not self.feed_urls: - raise ValueError("WireFeedConfig requires at least one feed URL") - if any(not url.strip() for url in self.feed_urls): - raise ValueError("feed URLs must not be empty") - - -@dataclass(frozen=True, slots=True) -class RawWireArtifact: - """Raw bytes and fetch metadata retained for replay and re-cleaning.""" - - bytes: bytes - content_hash: str - content_type: str - source_url: str | None - resolved_url: str | None - status_code: int | None - headers: dict[str, str] = field(default_factory=dict) - fetched_at: datetime | None = None - - -@dataclass(frozen=True, slots=True) -class WireDocument: - """Replayable raw wire evidence paired with cleaned markdown.""" - - provider: str - identity: str - cleaned_markdown: str - content_hash: str - raw_feed_entry: RawWireArtifact - raw_article: RawWireArtifact | None = None - payload_id: str | None = None - canonical_url: str | None = None - title: str | None = None - publisher: str | None = None - published_at: datetime | None = None - ticker_hints: tuple[NewsTickerHint, ...] = () - - -@dataclass(frozen=True, slots=True) -class WireFetchFailure: - """Lightweight record for one feed or item that could not be processed.""" - - provider: str - stage: WireFailureStage - source_url: str - item_id: str | None - error_type: str - message: str - - -@dataclass(frozen=True, slots=True) -class WireFetchResult: - """Successful wire documents plus non-fatal per-item failures.""" - - documents: tuple[WireDocument, ...] = () - failures: tuple[WireFetchFailure, ...] = () - - @property - def success_count(self) -> int: - """Number of documents produced by this call.""" - return len(self.documents) - - @property - def failure_count(self) -> int: - """Number of feed or item failures recorded by this call.""" - return len(self.failures) - - -async def fetch_wire_documents(config: WireFeedConfig) -> WireFetchResult: - """Fetch current feed items and convert them into wire documents. - - Independent feed and item failures are recorded while remaining inputs - continue. Configuration errors still raise before any network request. - """ - documents: list[WireDocument] = [] - failures: list[WireFetchFailure] = [] - seen: set[str] = set() - async with HttpFetcher(policy=config.fetch_policy) as fetcher: - for feed_url in config.feed_urls: - try: - fetched_feed = await fetcher.fetch_url( - feed_url, - max_bytes=5_000_000, - ) - except Exception as exc: - failures.append( - _failure( - config.provider, - "feed_fetch", - feed_url, - None, - exc, - ) - ) - continue - - try: - feed = parse_feed( - fetched_feed.bytes, - feed_url=feed_url, - content_type=fetched_feed.content_type, - headers=fetched_feed.headers, - fetched=fetched_feed, - ) - except Exception as exc: - failures.append( - _failure( - config.provider, - "feed_parse", - feed_url, - None, - exc, - ) - ) - continue - - for item in feed.items: - try: - mapping = config.provider.map_item(item) - identity = build_wire_identity( - provider=config.provider.name, - payload_id=mapping.payload_id, - canonical_url=mapping.canonical_url, - ) - except Exception as exc: - failures.append( - _failure( - config.provider, - "document_build", - item.url or feed_url, - item.id, - exc, - ) - ) - continue - - if identity in seen: - continue - seen.add(identity) - - document = await _build_document( - provider=config.provider, - item=item, - mapping=mapping, - identity=identity, - fetched_feed=fetched_feed, - fetcher=fetcher, - failures=failures, - ) - if document is not None: - documents.append(document) - - return WireFetchResult( - documents=tuple(documents), - failures=tuple(failures), - ) - - -def build_wire_identity( - *, - provider: str, - payload_id: str | None, - canonical_url: str | None, -) -> str: - """Build stable identity from provider, payload ID, and canonical URL.""" - normalized_provider = provider.strip().lower() - normalized_payload = (payload_id or "").strip() - normalized_url = ( - canonicalize_source_url(canonical_url) if canonical_url else "" - ) - if not normalized_provider: - raise ValueError("wire identity requires a provider") - if not normalized_payload and not normalized_url: - raise ValueError("wire identity requires a payload ID or URL") - raw_identity = "\x1f".join( - (normalized_provider, normalized_payload, normalized_url) - ) - digest = hashlib.sha256(raw_identity.encode("utf-8")).hexdigest() - return f"wire:{normalized_provider}:{digest}" - - -async def _build_document( - *, - provider: WireProvider, - item: FeedItem, - mapping: WireItemMapping, - identity: str, - fetched_feed: Fetched, - fetcher: HttpFetcher, - failures: list[WireFetchFailure], -) -> WireDocument | None: - raw_article: Fetched | None = None - if provider.body_source == "article": - article_url = mapping.canonical_url - if not article_url: - failures.append( - _failure( - provider, - "article_fetch", - item.source_feed_url or "", - item.id, - ValueError("article body source requires an item URL"), - ) - ) - return None - try: - raw_article = await fetcher.fetch_url( - article_url, - max_bytes=10_000_000, - ) - except Exception as exc: - failures.append( - _failure( - provider, - "article_fetch", - article_url, - item.id, - exc, - ) - ) - return None - - try: - if raw_article is None: - raw_news = await feed_item_to_news_document( - item, - publisher=provider.publisher, - body_source="feed", - ) - else: - raw_news = await news_document_from_fetched( - raw_article, - title=mapping.title, - publisher=provider.publisher, - published_at=mapping.published_at, - payload_id=mapping.payload_id, - ) - candidate = preprocess_news_document(raw_news) - except Exception as exc: - failures.append( - _failure( - provider, - "article_parse" - if raw_article is not None - else "document_build", - item.url or item.source_feed_url or "", - item.id, - exc, - ) - ) - return None - - return WireDocument( - provider=provider.name, - identity=identity, - cleaned_markdown=candidate.body_text, - content_hash=candidate.content_hash, - raw_feed_entry=_feed_entry_artifact(item, fetched_feed), - raw_article=( - _artifact_from_fetched(raw_article) - if raw_article is not None - else None - ), - payload_id=mapping.payload_id, - canonical_url=mapping.canonical_url, - title=candidate.title, - publisher=candidate.publisher, - published_at=candidate.published_at, - ticker_hints=candidate.ticker_hints, - ) - - -def _feed_entry_artifact( - item: FeedItem, - fetched_feed: Fetched, -) -> RawWireArtifact: - raw_xml = item.raw_xml - return RawWireArtifact( - bytes=raw_xml, - content_hash=hashlib.sha256(raw_xml).hexdigest(), - content_type="application/xml", - source_url=fetched_feed.source_url, - resolved_url=fetched_feed.resolved_url, - status_code=fetched_feed.status_code, - headers=dict(fetched_feed.headers), - fetched_at=fetched_feed.fetched_at, - ) - - -def _artifact_from_fetched(fetched: Fetched) -> RawWireArtifact: - return RawWireArtifact( - bytes=fetched.bytes, - content_hash=hashlib.sha256(fetched.bytes).hexdigest(), - content_type=fetched.content_type, - source_url=fetched.source_url, - resolved_url=fetched.resolved_url, - status_code=fetched.status_code, - headers=dict(fetched.headers), - fetched_at=fetched.fetched_at, - ) - - -def _failure( - provider: WireProvider, - stage: WireFailureStage, - source_url: str, - item_id: str | None, - error: Exception, -) -> WireFetchFailure: - return WireFetchFailure( - provider=provider.name, - stage=stage, - source_url=source_url, - item_id=item_id, - error_type=_error_type(error), - message=str(error), - ) - - -def _error_type(error: Exception) -> str: - if isinstance(error, FetchAttemptsExhausted): - return "retry_exhausted" - if isinstance(error, httpx.TimeoutException): - return "timeout" - if isinstance(error, httpx.TransportError): - return "network" - if isinstance(error, httpx.HTTPStatusError): - return "http_status" - if isinstance(error, ValueError): - return "invalid_content" - return "unexpected" diff --git a/scripts/smoke_wire.py b/scripts/smoke_wire.py deleted file mode 100644 index 92197f4..0000000 --- a/scripts/smoke_wire.py +++ /dev/null @@ -1,73 +0,0 @@ -#!/usr/bin/env python3 -"""Run the manual live-network checklist for PR Newswire.""" - -import asyncio - -from quantmind.preprocess import ( - PR_NEWSWIRE, - FetchPolicy, - WireFeedConfig, - fetch_wire_documents, -) - -_PUBLIC_FEED = "https://www.prnewswire.com/rss/news-releases-list.rss" - - -async def _check_pr_newswire() -> bool: - provider_name = PR_NEWSWIRE.name - try: - result = await asyncio.wait_for( - fetch_wire_documents( - WireFeedConfig( - provider=PR_NEWSWIRE, - feed_urls=(_PUBLIC_FEED,), - fetch_policy=FetchPolicy( - max_attempts=2, - min_interval_seconds=0.25, - ), - ) - ), - timeout=120, - ) - except Exception as exc: - print(f"[FAIL] {provider_name}: {type(exc).__name__}: {exc}") - return False - - documents = result.documents - checks = { - "documents": bool(documents), - "no_failures": not result.failures, - "raw_feed_entry": bool(documents) - and all(document.raw_feed_entry.bytes for document in documents), - "raw_article": bool(documents) - and all( - document.raw_article is not None - and bool(document.raw_article.bytes) - for document in documents - ), - "cleaned_markdown": bool(documents) - and all(document.cleaned_markdown.strip() for document in documents), - "unique_identity": len({document.identity for document in documents}) - == len(documents), - } - for check, passed in checks.items(): - state = "PASS" if passed else "FAIL" - print(f"[{state}] {provider_name}: {check}") - print( - f" documents={result.success_count} " - f"failures={result.failure_count}" - ) - for failure in result.failures[:3]: - print( - f" {failure.stage} {failure.error_type} {failure.source_url}" - ) - return all(checks.values()) - - -async def main() -> int: - """Run the PR Newswire checklist and return a process exit code.""" - return 0 if await _check_pr_newswire() else 1 - - -if __name__ == "__main__": - raise SystemExit(asyncio.run(main())) diff --git a/scripts/verify_news_e2e.py b/scripts/verify_news_e2e.py new file mode 100644 index 0000000..76a27e7 --- /dev/null +++ b/scripts/verify_news_e2e.py @@ -0,0 +1,101 @@ +#!/usr/bin/env python3 +"""Run the bounded live-network checks for public news collection.""" + +import asyncio +from datetime import datetime, timedelta, timezone + +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 + +_PUBLIC_RSS_URL = "https://www.prnewswire.com/rss/news-releases-list.rss" +_RSS_TIMEOUT_SECONDS = 60 +_DISCOVERY_TIMEOUT_SECONDS = 300 + + +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, + ), + timeout=30.0, + max_bytes=5_000_000, + ) as fetcher: + feed = await asyncio.wait_for( + fetch_rss_feed(_PUBLIC_RSS_URL, fetcher=fetcher), + timeout=_RSS_TIMEOUT_SECONDS, + ) + except Exception as exc: + print(f"[FAIL] rss: {type(exc).__name__}: {exc}") + return False + + usable_count = sum( + bool(item.title.strip() and (item.url or "").strip()) + for item in feed.items + ) + passed = bool(feed.items) and usable_count > 0 + state = "PASS" if passed else "FAIL" + print( + f"[{state}] rss: items={len(feed.items)} usable={usable_count} " + f"url={_PUBLIC_RSS_URL}" + ) + return passed + + +async def _check_discovery(start: datetime, end: datetime) -> bool: + 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 + + urls = [observation.canonical_url for observation in result.observations] + duplicate_count = len(urls) - len(set(urls)) + in_window_count = sum( + start <= observation.published_at < end + for observation in result.observations + ) + passed = ( + bool(result.observations) + and in_window_count == len(result.observations) + and result.complete + and not result.failures + ) + state = "PASS" if passed else "FAIL" + print( + f"[{state}] pr-newswire-discovery: " + f"observed={len(result.observations)} " + f"in_window={in_window_count} " + f"duplicates={duplicate_count} pages={result.page_count} " + f"failures={len(result.failures)} complete={result.complete}" + ) + for failure in result.failures[:3]: + print( + f" {failure.stage} {failure.error_type} {failure.source_url}" + ) + return passed + + +async def main(*, now: datetime | None = None) -> int: + """Run all live checks and return a process exit code.""" + end = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) + end = end.replace(second=0, microsecond=0) + start = end - timedelta(days=1) + 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 + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main())) diff --git a/tests/configs/test_base.py b/tests/configs/test_base.py index 395604c..164e237 100644 --- a/tests/configs/test_base.py +++ b/tests/configs/test_base.py @@ -50,12 +50,12 @@ def test_top_level_imports(self): ) from quantmind.configs import ( EarningsFlowCfg, - NewsFlowCfg, + NewsCollectionCfg, PaperFlowCfg, ) self.assertTrue(issubclass(PaperFlowCfg, BaseFlowCfgExport)) - self.assertTrue(issubclass(NewsFlowCfg, BaseFlowCfgExport)) + self.assertTrue(issubclass(NewsCollectionCfg, BaseFlowCfgExport)) self.assertTrue(issubclass(EarningsFlowCfg, BaseFlowCfgExport)) self.assertEqual(BaseInputExport.__name__, "BaseInput") diff --git a/tests/configs/test_news.py b/tests/configs/test_news.py index 33a2318..d2436b9 100644 --- a/tests/configs/test_news.py +++ b/tests/configs/test_news.py @@ -1,50 +1,69 @@ -"""Tests for configs.news.""" +"""Tests for the intent-oriented news configuration.""" import unittest +from datetime import datetime, timedelta, timezone -from pydantic import TypeAdapter, ValidationError +from pydantic import ValidationError -from quantmind.configs.news import ( - Headline, - HttpUrl, - NewsFlowCfg, - NewsInput, - RssFeed, -) +from quantmind.configs import NewsCollectionCfg, NewsWindow -class NewsFlowCfgTests(unittest.TestCase): - def test_defaults(self): - cfg = NewsFlowCfg() - self.assertEqual(cfg.model, "gpt-4o") - self.assertEqual(cfg.materiality_threshold, "medium") +class NewsCollectionCfgTests(unittest.TestCase): + def test_raw_html_is_not_retained_by_default(self) -> None: + self.assertFalse(NewsCollectionCfg().retain_raw_html) -class NewsInputTests(unittest.TestCase): - def setUp(self): - self.adapter = TypeAdapter(NewsInput) - - def test_rss(self): - v = self.adapter.validate_python( - {"type": "rss", "url": "https://feeds.example.com/markets"} +class NewsWindowTests(unittest.TestCase): + def test_normalizes_aware_timestamps_to_utc(self) -> None: + window = NewsWindow( + source="pr-newswire", + start=datetime( + 2026, + 7, + 13, + 8, + tzinfo=timezone(timedelta(hours=8)), + ), + end=datetime( + 2026, + 7, + 14, + 8, + tzinfo=timezone(timedelta(hours=8)), + ), ) - self.assertIsInstance(v, RssFeed) - def test_http(self): - v = self.adapter.validate_python( - {"type": "http", "url": "https://news.example.com/a"} - ) - self.assertIsInstance(v, HttpUrl) + self.assertEqual(window.start.tzinfo, timezone.utc) + self.assertEqual(window.start.hour, 0) + self.assertEqual(window.end.tzinfo, timezone.utc) - def test_headline(self): - v = self.adapter.validate_python( - {"type": "headline", "text": "Fed holds rates"} - ) - self.assertIsInstance(v, Headline) + def test_rejects_naive_timestamps(self) -> None: + with self.assertRaises(ValidationError): + NewsWindow( + source="pr-newswire", + start=datetime(2026, 7, 13), + end=datetime(2026, 7, 14, tzinfo=timezone.utc), + ) + + def test_rejects_empty_or_reversed_window(self) -> None: + timestamp = datetime(2026, 7, 14, tzinfo=timezone.utc) + + with self.assertRaisesRegex(ValidationError, "end to be after start"): + NewsWindow( + source="pr-newswire", + start=timestamp, + end=timestamp, + ) - def test_unknown_rejected(self): + def test_rejects_unsupported_source(self) -> None: with self.assertRaises(ValidationError): - self.adapter.validate_python({"type": "podcast", "url": "x"}) + NewsWindow.model_validate( + { + "source": "business-wire", + "start": datetime(2026, 7, 13, tzinfo=timezone.utc), + "end": datetime(2026, 7, 14, tzinfo=timezone.utc), + } + ) if __name__ == "__main__": diff --git a/tests/flows/test_news.py b/tests/flows/test_news.py new file mode 100644 index 0000000..314994f --- /dev/null +++ b/tests/flows/test_news.py @@ -0,0 +1,63 @@ +"""Tests for the agent-facing news collection operation.""" + +import unittest +from datetime import datetime, timezone +from unittest.mock import AsyncMock, patch + +from quantmind.configs import NewsCollectionCfg, NewsWindow +from quantmind.flows import collect_news +from quantmind.magic import _introspect_flow_signature +from quantmind.preprocess import NewsBatch + + +class CollectNewsTests(unittest.IsolatedAsyncioTestCase): + async def test_dispatches_pr_newswire_window( + self, + ) -> None: + window = NewsWindow( + source="pr-newswire", + start=datetime(2026, 7, 13, tzinfo=timezone.utc), + end=datetime(2026, 7, 14, tzinfo=timezone.utc), + ) + expected = NewsBatch(observed_count=3, complete=True) + + with patch( + "quantmind.flows.news._collect_pr_newswire", + new=AsyncMock(return_value=expected), + ) as collect: + result = await collect_news( + window, + cfg=NewsCollectionCfg(retain_raw_html=True), + ) + + self.assertIs(result, expected) + collect.assert_awaited_once_with( + start=window.start, + end=window.end, + retain_raw_html=True, + ) + + async def test_uses_agent_safe_retention_default(self) -> None: + window = NewsWindow( + source="pr-newswire", + start=datetime(2026, 7, 13, tzinfo=timezone.utc), + end=datetime(2026, 7, 14, tzinfo=timezone.utc), + ) + + with patch( + "quantmind.flows.news._collect_pr_newswire", + new=AsyncMock(return_value=NewsBatch(complete=True)), + ) as collect: + await collect_news(window) + + self.assertFalse(collect.await_args.kwargs["retain_raw_html"]) + + def test_signature_is_magic_input_compatible(self) -> None: + input_type, cfg_type = _introspect_flow_signature(collect_news) + + self.assertIs(input_type, NewsWindow) + self.assertIs(cfg_type, NewsCollectionCfg) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/preprocess/fixtures/pr_newswire/article.html b/tests/preprocess/fixtures/pr_newswire/article.html new file mode 100644 index 0000000..51464c5 --- /dev/null +++ b/tests/preprocess/fixtures/pr_newswire/article.html @@ -0,0 +1,10 @@ + + + +
+

Example Company Shares an Operations Update

+

Example Company (NYSE: EXCO) supplied fictional operating details.

+

This full article is the source of the cleaned Markdown body.

+
+ + diff --git a/tests/preprocess/fixtures/pr_newswire/listing_full.html b/tests/preprocess/fixtures/pr_newswire/listing_full.html new file mode 100644 index 0000000..138dda8 --- /dev/null +++ b/tests/preprocess/fixtures/pr_newswire/listing_full.html @@ -0,0 +1,34 @@ + + + + +
+
+ +

Jul 14, 2026, 00:30 ETEnd Boundary

+ +
+
+
+
+ +

Jul 14, 2026, 00:29 ETIn Window

+ +
+
+
+
+ +

Jul 13, 2026, 23:58 ETStart Boundary

+ +
+
+
+
+ +

Jul 13, 2026, 23:57 ETOlder Observation

+ +
+
+ + diff --git a/tests/preprocess/fixtures/pr_newswire/listing_short_page_1.html b/tests/preprocess/fixtures/pr_newswire/listing_short_page_1.html new file mode 100644 index 0000000..3baa1f9 --- /dev/null +++ b/tests/preprocess/fixtures/pr_newswire/listing_short_page_1.html @@ -0,0 +1,30 @@ + + + + +
+
+ +

00:10 ETExample Alpha & Partners

+

First sanitized teaser.

+ +
+
+
+
+ +

00:10 ETExample Alpha & Partners

+

Repeated source observation.

+ +
+
+
+
+ +

00:01 ETExample Beta

+

Last observation before midnight.

+ +
+
+ + diff --git a/tests/preprocess/fixtures/pr_newswire/listing_short_page_2.html b/tests/preprocess/fixtures/pr_newswire/listing_short_page_2.html new file mode 100644 index 0000000..27a10ac --- /dev/null +++ b/tests/preprocess/fixtures/pr_newswire/listing_short_page_2.html @@ -0,0 +1,22 @@ + + + + +
+
+ +

23:59 ETExample Gamma

+

The inferred date must roll back one day.

+ +
+
+
+
+ +

23:57 ETExample Delta

+

This row proves that discovery crossed the window start.

+ +
+
+ + diff --git a/tests/preprocess/fixtures/wire/pr_news_wire.xml b/tests/preprocess/fixtures/wire/pr_news_wire.xml deleted file mode 100644 index b493c9f..0000000 --- a/tests/preprocess/fixtures/wire/pr_news_wire.xml +++ /dev/null @@ -1,12 +0,0 @@ - - - Sanitized PR Newswire feed - - Example Energy Shares an Operations Update - https://example.test/prn/releases/operations-update - prn-example-001 - Wed, 17 Apr 2024 12:00:00 GMT - Example Energy shared a short operations teaser.

]]> - - - diff --git a/tests/preprocess/fixtures/wire/pr_news_wire_article.html b/tests/preprocess/fixtures/wire/pr_news_wire_article.html deleted file mode 100644 index e55dd02..0000000 --- a/tests/preprocess/fixtures/wire/pr_news_wire_article.html +++ /dev/null @@ -1,10 +0,0 @@ - - - -
-

Example Energy Shares an Operations Update

-

Example Energy (NYSE: EXEN) supplied fictional operating details for a sanitized provider fixture.

-

This full article is the source of the cleaned markdown body.

-
- - diff --git a/tests/preprocess/test_news.py b/tests/preprocess/test_news.py index a2755a6..6c04a8b 100644 --- a/tests/preprocess/test_news.py +++ b/tests/preprocess/test_news.py @@ -10,8 +10,8 @@ from quantmind.preprocess.fetch.rss import FeedItem from quantmind.preprocess.news import ( RawNewsDocument, - build_news_dedup_key, - build_sec_news_dedup_key, + build_news_identity, + build_sec_news_identity, canonicalize_source_url, extract_exchange_ticker_hints, feed_item_to_news_document, @@ -84,18 +84,18 @@ def test_exchange_ticker_hints_are_deduped(self): self.assertEqual(hints[0].exchange, "NASDAQ") self.assertEqual(hints[1].exchange, "NYSE") - def test_build_sec_news_dedup_key(self): + def test_build_sec_news_identity(self): self.assertEqual( - build_sec_news_dedup_key( + build_sec_news_identity( accession_number="0001045810-26-000123", section_key="EX99.1", ), "sec:0001045810-26-000123:ex99.1", ) - def test_build_news_dedup_key_requires_identity(self): + def test_build_news_identity_requires_source_reference(self): with self.assertRaises(ValueError): - build_news_dedup_key(source_type="press_release") + build_news_identity(source_type="press_release") def test_preprocess_news_document_builds_candidate_contract(self): published = datetime( @@ -116,7 +116,7 @@ def test_preprocess_news_document_builds_candidate_contract(self): candidate = preprocess_news_document(raw) self.assertEqual(candidate.source_type, "press_release") - self.assertTrue(candidate.dedup_key.startswith("wire:")) + self.assertTrue(candidate.identity.startswith("wire:")) self.assertEqual( candidate.source_url, "https://example.com/pr/nvidia-results" ) diff --git a/tests/preprocess/test_news_types.py b/tests/preprocess/test_news_types.py new file mode 100644 index 0000000..cfa889e --- /dev/null +++ b/tests/preprocess/test_news_types.py @@ -0,0 +1,43 @@ +"""Tests for generic news collection result contracts.""" + +import unittest +from dataclasses import FrozenInstanceError + +from quantmind.preprocess import NewsArtifact, NewsBatch, NewsFailure + + +class NewsResultContractTests(unittest.TestCase): + def test_batch_counts_documents_and_failures(self) -> None: + failure = NewsFailure( + source="pr-newswire", + stage="article_fetch", + source_url="https://example.test/release", + item_id="release-1", + error_type="timeout", + message="request timed out", + ) + batch = NewsBatch(failures=(failure,), observed_count=1, complete=True) + + self.assertEqual(batch.success_count, 0) + self.assertEqual(batch.failure_count, 1) + self.assertEqual(batch.observed_count, 1) + self.assertTrue(batch.complete) + + def test_artifact_can_retain_hash_without_bytes(self) -> None: + artifact = NewsArtifact( + bytes=None, + content_hash="abc123", + content_type="text/html", + source_url="https://example.test/release", + resolved_url="https://example.test/release", + status_code=200, + ) + + self.assertIsNone(artifact.bytes) + self.assertEqual(artifact.content_hash, "abc123") + with self.assertRaises(FrozenInstanceError): + setattr(artifact, "bytes", b"changed") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/preprocess/test_pr_newswire.py b/tests/preprocess/test_pr_newswire.py new file mode 100644 index 0000000..856b5f7 --- /dev/null +++ b/tests/preprocess/test_pr_newswire.py @@ -0,0 +1,437 @@ +import re +import unittest +from datetime import datetime, timezone +from pathlib import Path +from unittest.mock import patch + +import httpx +import respx + +from quantmind.preprocess.fetch.http import FetchPolicy +from quantmind.preprocess.pr_newswire import ( + _collect_pr_newswire, + _discover_pr_newswire, +) + +_FIXTURES = Path(__file__).parent / "fixtures" / "pr_newswire" +_LISTING_RE = re.compile( + r"https://www\.prnewswire\.com/news-releases/" + r"news-releases-list/\?.*" +) +_TEST_POLICY = FetchPolicy( + max_attempts=1, + backoff_base_seconds=0.0, + backoff_max_seconds=0.0, + jitter_seconds=0.0, + max_concurrency_per_host=2, + min_interval_seconds=0.0, +) + + +def _fixture(name: str) -> bytes: + return (_FIXTURES / name).read_bytes() + + +def _listing_response(content: bytes, request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + request=request, + headers={"Content-Type": "text/html"}, + content=content, + ) + + +def _short_listing(*rows: tuple[str, str]) -> bytes: + cards = "".join( + ( + '
" + ) + for timestamp, payload_id in rows + ) + return ( + '' + f"{cards}" + ).encode() + + +class DiscoverPRNewswireTests(unittest.IsolatedAsyncioTestCase): + async def test_preserves_duplicates_and_rolls_short_times_across_pages( + self, + ) -> None: + pages = { + 1: _fixture("listing_short_page_1.html"), + 2: _fixture("listing_short_page_2.html"), + } + + def respond(request: httpx.Request) -> httpx.Response: + page = int(request.url.params["page"]) + return _listing_response(pages[page], request) + + with ( + patch( + "quantmind.preprocess.pr_newswire._DEFAULT_FETCH_POLICY", + _TEST_POLICY, + ), + respx.mock(assert_all_called=True) as router, + ): + route = router.get(url__regex=_LISTING_RE).mock(side_effect=respond) + result = await _discover_pr_newswire( + start=datetime(2026, 7, 14, 3, 58, tzinfo=timezone.utc), + end=datetime(2026, 7, 14, 4, 30, tzinfo=timezone.utc), + ) + + self.assertTrue(result.complete) + self.assertEqual(result.page_count, 2) + self.assertEqual(result.observed_count, 4) + self.assertEqual(route.call_count, 2) + self.assertEqual( + [item.payload_id for item in result.observations], + ["302900111", "302900111", "302900112", "302900113"], + ) + self.assertEqual( + result.observations[-1].published_at, + datetime(2026, 7, 14, 3, 59, tzinfo=timezone.utc), + ) + self.assertEqual( + result.observations[0].identity, + result.observations[1].identity, + ) + self.assertNotEqual( + result.observations[0].canonical_url, + result.observations[1].canonical_url, + ) + self.assertEqual( + len({item.identity for item in result.observations}), 3 + ) + artifact = result.observations[0].discovery_artifact + self.assertIn( + b"Repeated", result.observations[1].discovery_artifact.bytes + ) + self.assertIn(b"Example Alpha", artifact.bytes) + self.assertEqual(artifact.content_type, "text/html") + self.assertEqual(artifact.status_code, 200) + request = route.calls[0].request + self.assertEqual(request.url.params["hour"], "01") + + async def test_full_timestamps_use_half_open_window_boundaries( + self, + ) -> None: + with ( + patch( + "quantmind.preprocess.pr_newswire._DEFAULT_FETCH_POLICY", + _TEST_POLICY, + ), + respx.mock(assert_all_called=True) as router, + ): + router.get(url__regex=_LISTING_RE).mock( + side_effect=lambda request: _listing_response( + _fixture("listing_full.html"), request + ) + ) + result = await _discover_pr_newswire( + start=datetime(2026, 7, 14, 3, 58, tzinfo=timezone.utc), + end=datetime(2026, 7, 14, 4, 30, tzinfo=timezone.utc), + ) + + self.assertTrue(result.complete) + self.assertEqual( + [item.payload_id for item in result.observations], + ["302900202", "302900203"], + ) + self.assertEqual( + result.observations[-1].published_at, + datetime(2026, 7, 14, 3, 58, tzinfo=timezone.utc), + ) + + async def test_equal_start_cluster_continues_onto_the_next_page( + self, + ) -> None: + pages = { + 1: _short_listing( + ("00:10 ET", "302900401"), + ("00:00 ET", "302900402"), + ), + 2: _short_listing( + ("00:00 ET", "302900403"), + ("23:59 ET", "302900404"), + ), + } + + def respond(request: httpx.Request) -> httpx.Response: + return _listing_response( + pages[int(request.url.params["page"])], + request, + ) + + with ( + patch( + "quantmind.preprocess.pr_newswire._DEFAULT_FETCH_POLICY", + _TEST_POLICY, + ), + respx.mock(assert_all_called=True) as router, + ): + router.get(url__regex=_LISTING_RE).mock(side_effect=respond) + result = await _discover_pr_newswire( + start=datetime(2026, 7, 14, 4, 0, tzinfo=timezone.utc), + end=datetime(2026, 7, 14, 4, 30, tzinfo=timezone.utc), + ) + + self.assertTrue(result.complete) + self.assertEqual(result.page_count, 2) + self.assertEqual( + [item.payload_id for item in result.observations], + ["302900401", "302900402", "302900403"], + ) + + async def test_row_parse_failure_keeps_good_rows_but_is_incomplete( + self, + ) -> None: + listing = b""" + +
+
+
+ + """ + with ( + patch( + "quantmind.preprocess.pr_newswire._DEFAULT_FETCH_POLICY", + _TEST_POLICY, + ), + respx.mock(assert_all_called=True) as router, + ): + router.get(url__regex=_LISTING_RE).mock( + side_effect=lambda request: _listing_response(listing, request) + ) + result = await _discover_pr_newswire( + start=datetime(2026, 7, 14, 3, 58, tzinfo=timezone.utc), + end=datetime(2026, 7, 14, 4, 30, tzinfo=timezone.utc), + ) + + self.assertFalse(result.complete) + self.assertEqual(result.observed_count, 1) + self.assertEqual(result.failures[0].stage, "discovery_parse") + self.assertEqual(result.failures[0].error_type, "invalid_content") + + async def test_fetch_failure_is_recorded(self) -> None: + with ( + patch( + "quantmind.preprocess.pr_newswire._DEFAULT_FETCH_POLICY", + _TEST_POLICY, + ), + respx.mock(assert_all_called=True) as router, + ): + router.get(url__regex=_LISTING_RE).mock( + return_value=httpx.Response(400) + ) + result = await _discover_pr_newswire( + start=datetime(2026, 7, 14, 3, 58, tzinfo=timezone.utc), + end=datetime(2026, 7, 14, 4, 30, tzinfo=timezone.utc), + ) + + self.assertFalse(result.complete) + self.assertEqual(result.page_count, 0) + self.assertEqual(result.failures[0].stage, "discovery_fetch") + self.assertEqual(result.failures[0].error_type, "http_status") + + async def test_page_limit_is_an_explicit_incomplete_failure(self) -> None: + pages = { + 1: _fixture("listing_short_page_1.html"), + 2: _fixture("listing_short_page_2.html"), + } + + def respond(request: httpx.Request) -> httpx.Response: + return _listing_response( + pages[int(request.url.params["page"])], + request, + ) + + with ( + patch( + "quantmind.preprocess.pr_newswire._DEFAULT_FETCH_POLICY", + _TEST_POLICY, + ), + patch("quantmind.preprocess.pr_newswire._MAX_PAGES", 2), + respx.mock(assert_all_called=True) as router, + ): + router.get(url__regex=_LISTING_RE).mock(side_effect=respond) + result = await _discover_pr_newswire( + start=datetime(2026, 7, 12, tzinfo=timezone.utc), + end=datetime(2026, 7, 14, 4, 30, tzinfo=timezone.utc), + ) + + self.assertFalse(result.complete) + self.assertEqual(result.page_count, 2) + self.assertIn("exceeded 2 pages", result.failures[0].message) + + async def test_cache_bust_retries_listing_404(self) -> None: + with ( + patch( + "quantmind.preprocess.pr_newswire._DEFAULT_FETCH_POLICY", + _TEST_POLICY, + ), + respx.mock(assert_all_called=True) as router, + ): + route = router.get(url__regex=_LISTING_RE).mock( + side_effect=[ + httpx.Response(404), + httpx.Response( + 200, + headers={"Content-Type": "text/html"}, + content=_fixture("listing_full.html"), + ), + ] + ) + result = await _discover_pr_newswire( + start=datetime(2026, 7, 14, 4, 29, tzinfo=timezone.utc), + end=datetime(2026, 7, 14, 4, 30, tzinfo=timezone.utc), + ) + + self.assertTrue(result.complete) + self.assertEqual(route.call_count, 2) + self.assertNotEqual( + route.calls[0].request.url.params["_"], + route.calls[1].request.url.params["_"], + ) + + async def test_rejects_invalid_windows_before_network_work(self) -> None: + with self.assertRaisesRegex(ValueError, "aware start"): + await _discover_pr_newswire( + start=datetime(2026, 7, 14, 3, 58), + end=datetime(2026, 7, 14, 4, 30, tzinfo=timezone.utc), + ) + with self.assertRaisesRegex(ValueError, "end after start"): + await _discover_pr_newswire( + start=datetime(2026, 7, 14, 4, 30, tzinfo=timezone.utc), + end=datetime(2026, 7, 14, 4, 30, tzinfo=timezone.utc), + ) + + +class CollectPRNewswireTests(unittest.IsolatedAsyncioTestCase): + async def test_collects_article_and_discards_raw_html_by_default( + self, + ) -> None: + article_url = ( + "https://www.prnewswire.com/news-releases/in-window-302900202.html" + ) + with ( + patch( + "quantmind.preprocess.pr_newswire._DEFAULT_FETCH_POLICY", + _TEST_POLICY, + ), + respx.mock(assert_all_called=True) as router, + ): + router.get(url__regex=_LISTING_RE).mock( + side_effect=lambda request: _listing_response( + _fixture("listing_full.html"), request + ) + ) + router.get(article_url).mock( + return_value=httpx.Response( + 200, + headers={"Content-Type": "text/html"}, + content=_fixture("article.html"), + ) + ) + result = await _collect_pr_newswire( + start=datetime(2026, 7, 14, 4, 29, tzinfo=timezone.utc), + end=datetime(2026, 7, 14, 4, 30, tzinfo=timezone.utc), + retain_raw_html=False, + ) + + self.assertTrue(result.complete) + self.assertEqual(result.observed_count, 1) + self.assertEqual(result.success_count, 1) + self.assertEqual(result.failure_count, 0) + document = result.documents[0] + self.assertIn("fictional operating details", document.cleaned_markdown) + self.assertIsNone(document.article_artifact.bytes) + self.assertTrue(document.article_artifact.content_hash) + self.assertTrue(document.discovery_artifact.bytes) + self.assertEqual( + document.identity.split(":", 2)[:2], ["news", "pr-newswire"] + ) + self.assertEqual(document.ticker_hints[0].symbol, "EXCO") + + async def test_retain_raw_html_and_record_independent_article_failure( + self, + ) -> None: + article_url = ( + "https://www.prnewswire.com/news-releases/in-window-302900202.html" + ) + with ( + patch( + "quantmind.preprocess.pr_newswire._DEFAULT_FETCH_POLICY", + _TEST_POLICY, + ), + respx.mock(assert_all_called=True) as router, + ): + router.get(url__regex=_LISTING_RE).mock( + side_effect=lambda request: _listing_response( + _fixture("listing_full.html"), request + ) + ) + router.get(article_url).mock(return_value=httpx.Response(404)) + failed = await _collect_pr_newswire( + start=datetime(2026, 7, 14, 4, 29, tzinfo=timezone.utc), + end=datetime(2026, 7, 14, 4, 30, tzinfo=timezone.utc), + retain_raw_html=True, + ) + + self.assertTrue(failed.complete) + self.assertEqual(failed.observed_count, 1) + self.assertEqual(failed.success_count, 0) + self.assertEqual(failed.failures[0].stage, "article_fetch") + + with ( + patch( + "quantmind.preprocess.pr_newswire._DEFAULT_FETCH_POLICY", + _TEST_POLICY, + ), + respx.mock(assert_all_called=True) as router, + ): + router.get(url__regex=_LISTING_RE).mock( + side_effect=lambda request: _listing_response( + _fixture("listing_full.html"), request + ) + ) + router.get(article_url).mock( + return_value=httpx.Response( + 200, + headers={"Content-Type": "text/html"}, + content=_fixture("article.html"), + ) + ) + retained = await _collect_pr_newswire( + start=datetime(2026, 7, 14, 4, 29, tzinfo=timezone.utc), + end=datetime(2026, 7, 14, 4, 30, tzinfo=timezone.utc), + retain_raw_html=True, + ) + + self.assertEqual( + retained.documents[0].article_artifact.bytes, + _fixture("article.html"), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/preprocess/test_wire.py b/tests/preprocess/test_wire.py deleted file mode 100644 index 44563c2..0000000 --- a/tests/preprocess/test_wire.py +++ /dev/null @@ -1,276 +0,0 @@ -"""Tests for provider-pluggable wire ingestion.""" - -import unittest -from pathlib import Path - -import httpx -import respx - -from quantmind.preprocess.fetch.http import FetchPolicy -from quantmind.preprocess.wire import ( - PR_NEWSWIRE, - WireFeedConfig, - WireItemMapping, - build_wire_identity, - fetch_wire_documents, -) - -_FIXTURES = Path(__file__).parent / "fixtures" / "wire" - - -def _fixture(name: str) -> bytes: - return (_FIXTURES / name).read_bytes() - - -class _FeedBodyProvider: - name = "full-content-test-wire" - publisher = "Full Content Test Wire" - body_source = "feed" - - def map_item(self, item): - return WireItemMapping( - payload_id=item.id, - canonical_url=item.url, - title=item.title, - published_at=item.published_at, - ) - - -class WireProviderConfigTests(unittest.TestCase): - def test_builtin_body_sources_are_explicit(self): - self.assertEqual(PR_NEWSWIRE.body_source, "article") - - def test_config_requires_a_feed_url(self): - with self.assertRaises(ValueError): - WireFeedConfig(provider=PR_NEWSWIRE, feed_urls=()) - - def test_identity_uses_all_available_components(self): - base = build_wire_identity( - provider="provider-a", - payload_id="item-1", - canonical_url="https://example.test/release", - ) - - self.assertNotEqual( - base, - build_wire_identity( - provider="provider-b", - payload_id="item-1", - canonical_url="https://example.test/release", - ), - ) - self.assertNotEqual( - base, - build_wire_identity( - provider="provider-a", - payload_id="item-2", - canonical_url="https://example.test/release", - ), - ) - self.assertNotEqual( - base, - build_wire_identity( - provider="provider-a", - payload_id="item-1", - canonical_url="https://example.test/other", - ), - ) - - -class FetchWireDocumentsTests(unittest.IsolatedAsyncioTestCase): - async def test_feed_body_mode_remains_available_for_custom_provider(self): - feed_url = "https://example.test/custom/rss" - feed = b""" - - Example quarterly results - https://example.test/releases/results - custom-example-001 - Example Corp reported fictional quarterly revenue.

-

NASDAQ: EXMPL

- ]]> - - """ - with respx.mock(assert_all_called=True) as router: - router.get(feed_url).mock( - return_value=httpx.Response( - 200, - headers={"Content-Type": "application/rss+xml"}, - content=feed, - ) - ) - result = await fetch_wire_documents( - WireFeedConfig( - provider=_FeedBodyProvider(), - feed_urls=(feed_url,), - ) - ) - - self.assertEqual(result.success_count, 1) - self.assertEqual(result.failure_count, 0) - document = result.documents[0] - self.assertIn("fictional quarterly revenue", document.cleaned_markdown) - self.assertIsNone(document.raw_article) - self.assertIn(b"custom-example-001", document.raw_feed_entry.bytes) - self.assertEqual(document.raw_feed_entry.status_code, 200) - self.assertIsNotNone(document.raw_feed_entry.fetched_at) - self.assertEqual(document.ticker_hints[0].symbol, "EXMPL") - - async def test_pr_newswire_returns_both_raw_artifacts(self): - feed_url = "https://example.test/prn/rss" - article_url = "https://example.test/prn/releases/operations-update" - article = _fixture("pr_news_wire_article.html") - with respx.mock(assert_all_called=True) as router: - router.get(feed_url).mock( - return_value=httpx.Response( - 200, - headers={"Content-Type": "application/rss+xml"}, - content=_fixture("pr_news_wire.xml"), - ) - ) - router.get(article_url).mock( - return_value=httpx.Response( - 200, - headers={"Content-Type": "text/html"}, - content=article, - ) - ) - result = await fetch_wire_documents( - WireFeedConfig( - provider=PR_NEWSWIRE, - feed_urls=(feed_url,), - ) - ) - - self.assertEqual(result.success_count, 1) - self.assertEqual(result.failure_count, 0) - document = result.documents[0] - self.assertIn("full article", document.cleaned_markdown) - self.assertIsNotNone(document.raw_article) - assert document.raw_article is not None - self.assertEqual(document.raw_article.bytes, article) - self.assertTrue(document.raw_feed_entry.bytes) - - async def test_duplicate_identity_is_processed_once(self): - feed_url = "https://example.test/duplicate/rss" - item = """ - - Duplicate release - https://example.test/releases/duplicate - duplicate-1 - A complete duplicate body.

]]> - - """ - feed = f"{item}{item}".encode() - with respx.mock(assert_all_called=True) as router: - router.get(feed_url).mock( - return_value=httpx.Response( - 200, - headers={"Content-Type": "application/rss+xml"}, - content=feed, - ) - ) - result = await fetch_wire_documents( - WireFeedConfig( - provider=_FeedBodyProvider(), - feed_urls=(feed_url,), - ) - ) - - self.assertEqual(result.success_count, 1) - self.assertEqual(result.failure_count, 0) - - async def test_extraction_failure_is_recorded_and_batch_continues(self): - feed_url = "https://example.test/partial/rss" - bad_url = "https://example.test/releases/bad" - good_url = "https://example.test/releases/good" - feed = f""" - - - Bad release{bad_url} - bad-1Short teaser - - - Good release{good_url} - good-1Short teaser - - - """.encode() - good_article = b""" -
-

Good release

The complete good article body.

-
- """ - with respx.mock(assert_all_called=True) as router: - router.get(feed_url).mock( - return_value=httpx.Response( - 200, - headers={"Content-Type": "application/rss+xml"}, - content=feed, - ) - ) - router.get(bad_url).mock( - return_value=httpx.Response( - 200, - headers={"Content-Type": "application/pdf"}, - content=b"not html", - ) - ) - router.get(good_url).mock( - return_value=httpx.Response( - 200, - headers={"Content-Type": "text/html"}, - content=good_article, - ) - ) - result = await fetch_wire_documents( - WireFeedConfig( - provider=PR_NEWSWIRE, - feed_urls=(feed_url,), - fetch_policy=FetchPolicy(max_attempts=1), - ) - ) - - self.assertEqual(result.success_count, 1) - self.assertEqual(result.failure_count, 1) - self.assertEqual(result.documents[0].payload_id, "good-1") - failure = result.failures[0] - self.assertEqual(failure.item_id, "bad-1") - self.assertEqual(failure.stage, "article_parse") - self.assertEqual(failure.error_type, "invalid_content") - - async def test_feed_failure_does_not_discard_other_feed(self): - bad_feed = "https://example.test/feed/bad" - good_feed = "https://example.test/feed/good" - with respx.mock(assert_all_called=True) as router: - router.get(bad_feed).mock(return_value=httpx.Response(404)) - router.get(good_feed).mock( - return_value=httpx.Response( - 200, - headers={"Content-Type": "application/rss+xml"}, - content=b""" - - Good feed item - https://example.test/releases/good-feed - good-feed-1 - A complete feed body. - - """, - ) - ) - result = await fetch_wire_documents( - WireFeedConfig( - provider=_FeedBodyProvider(), - feed_urls=(bad_feed, good_feed), - fetch_policy=FetchPolicy(max_attempts=1), - ) - ) - - self.assertEqual(result.success_count, 1) - self.assertEqual(result.failure_count, 1) - self.assertEqual(result.failures[0].stage, "feed_fetch") - self.assertEqual(result.failures[0].error_type, "http_status") - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_magic.py b/tests/test_magic.py index 426f6cd..80eba39 100644 --- a/tests/test_magic.py +++ b/tests/test_magic.py @@ -44,6 +44,19 @@ def test_paper_flow_returns_paper_input_and_cfg(self) -> None: # PaperInput is the Annotated[Union[...]] alias; pass through. self.assertEqual(input_type, PaperInput) + def test_resolves_postponed_annotations(self) -> None: + async def postponed_flow( + input: "ArxivIdentifier", + *, + cfg: "PaperFlowCfg | None" = None, + ) -> None: + return None + + input_type, cfg_type = _introspect_flow_signature(postponed_flow) + + self.assertIs(input_type, ArxivIdentifier) + self.assertIs(cfg_type, PaperFlowCfg) + def test_missing_input_param_raises(self) -> None: async def bad(*, cfg: PaperFlowCfg | None = None) -> None: return None diff --git a/tests/test_verify_news_e2e.py b/tests/test_verify_news_e2e.py new file mode 100644 index 0000000..ca5d684 --- /dev/null +++ b/tests/test_verify_news_e2e.py @@ -0,0 +1,142 @@ +import io +import unittest +from contextlib import redirect_stdout +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +from scripts import verify_news_e2e + +_PUBLISHED_AT = datetime(2026, 7, 14, tzinfo=timezone.utc) + + +def _feed_item(*, title: str = "Release", url: str | None = "https://x"): + return SimpleNamespace(title=title, url=url) + + +def _discovery_result( + *, + urls: tuple[str, ...] = ("https://example.test/release",), + failures: tuple[object, ...] = (), + complete: bool = True, + published_at: datetime = _PUBLISHED_AT, +): + return SimpleNamespace( + observations=tuple( + SimpleNamespace( + canonical_url=url, + published_at=published_at, + ) + for url in urls + ), + failures=failures, + complete=complete, + page_count=2, + ) + + +class VerifyNewsE2ETests(unittest.IsolatedAsyncioTestCase): + async def test_main_reports_duplicates_and_passes_both_components(self): + now = datetime(2026, 7, 14, 12, 30, tzinfo=timezone.utc) + feed = SimpleNamespace(items=(_feed_item(), _feed_item(url=None))) + discovery = _discovery_result( + urls=( + "https://example.test/releases/one", + "https://example.test/releases/one", + "https://example.test/releases/two", + ) + ) + + with ( + patch.object( + verify_news_e2e, + "fetch_rss_feed", + new=AsyncMock(return_value=feed), + ), + patch.object( + verify_news_e2e, + "_discover_pr_newswire", + new=AsyncMock(return_value=discovery), + ) as discover, + redirect_stdout(io.StringIO()) as output, + ): + exit_code = await verify_news_e2e.main(now=now) + + self.assertEqual(exit_code, 0) + self.assertIn("[PASS] rss", output.getvalue()) + self.assertIn("duplicates=1", output.getvalue()) + discover.assert_awaited_once_with( + start=now - timedelta(days=1), + end=now, + ) + + async def test_rss_failure_does_not_skip_discovery(self): + discovery = _discovery_result() + with ( + patch.object( + verify_news_e2e, + "fetch_rss_feed", + new=AsyncMock(side_effect=ValueError("bad XML")), + ), + patch.object( + verify_news_e2e, + "_discover_pr_newswire", + new=AsyncMock(return_value=discovery), + ) as discover, + redirect_stdout(io.StringIO()) as output, + ): + exit_code = await verify_news_e2e.main() + + self.assertEqual(exit_code, 1) + self.assertIn("[FAIL] rss: ValueError: bad XML", output.getvalue()) + discover.assert_awaited_once() + + async def test_rss_rejects_empty_or_wholly_unusable_items(self): + feeds = ( + SimpleNamespace(items=()), + SimpleNamespace(items=(_feed_item(url=None),)), + ) + for feed in feeds: + with self.subTest(feed=feed): + with ( + patch.object( + verify_news_e2e, + "fetch_rss_feed", + new=AsyncMock(return_value=feed), + ), + redirect_stdout(io.StringIO()), + ): + passed = await verify_news_e2e._check_rss() + self.assertFalse(passed) + + async def test_discovery_rejects_invalid_results(self): + start = datetime(2026, 7, 13, tzinfo=timezone.utc) + end = start + timedelta(days=1) + failure = SimpleNamespace( + stage="discovery_fetch", + error_type="http_status", + source_url="https://example.test/list", + ) + results = ( + _discovery_result(urls=()), + _discovery_result(complete=False), + _discovery_result(failures=(failure,)), + _discovery_result(published_at=end), + ) + + for result in results: + with self.subTest(result=result): + with ( + patch.object( + verify_news_e2e, + "_discover_pr_newswire", + new=AsyncMock(return_value=result), + ), + redirect_stdout(io.StringIO()), + ): + passed = await verify_news_e2e._check_discovery(start, end) + self.assertFalse(passed) + + +if __name__ == "__main__": + unittest.main()