diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3a2d0fe..6207fd4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -28,6 +28,17 @@ jobs: name: ruff-diagnostics path: ruff-output.txt retention-days: 7 + - name: Ruff format + run: | + set -o pipefail + uv run ruff format --check --diff src tests 2>&1 | tee ruff-format-output.txt + - name: Upload Ruff format diagnostics + if: failure() + uses: actions/upload-artifact@v4 + with: + name: ruff-format-diagnostics + path: ruff-format-output.txt + retention-days: 7 - name: Mypy run: uv run mypy src - name: Pytest diff --git a/dbt/models/marts/fct_financial_facts.sql b/dbt/models/marts/fct_financial_facts.sql index 88bbde7..9144e1c 100644 --- a/dbt/models/marts/fct_financial_facts.sql +++ b/dbt/models/marts/fct_financial_facts.sql @@ -6,15 +6,32 @@ deduplicated as ( select *, row_number() over ( - partition by cik, fact_namespace, fact_name, unit, accession_number, period_end_date + partition by + cik, + fact_namespace, + fact_name, + unit, + accession_number, + period_start_date, + period_end_date order by filed_date desc ) as row_number from facts ) select - md5(concat_ws('|', cik, fact_namespace, fact_name, unit, accession_number, period_end_date)) - as financial_fact_id, + md5( + concat_ws( + '|', + cik, + fact_namespace, + fact_name, + unit, + accession_number, + coalesce(cast(period_start_date as varchar), '__instant__'), + cast(period_end_date as varchar) + ) + ) as financial_fact_id, cik, entity_name, fact_namespace, diff --git a/dbt/tests/test_fct_financial_facts_preserves_period_grain.sql b/dbt/tests/test_fct_financial_facts_preserves_period_grain.sql new file mode 100644 index 0000000..e8dce2a --- /dev/null +++ b/dbt/tests/test_fct_financial_facts_preserves_period_grain.sql @@ -0,0 +1,39 @@ +with source_periods as ( + select distinct + cik, + fact_namespace, + fact_name, + unit, + accession_number, + period_end_date, + coalesce(cast(period_start_date as varchar), '__instant__') as period_start_key + from {{ ref('stg_financial_facts') }} +), + +mart_periods as ( + select distinct + cik, + fact_namespace, + fact_name, + unit, + accession_number, + period_end_date, + coalesce(cast(period_start_date as varchar), '__instant__') as period_start_key + from {{ ref('fct_financial_facts') }} +), + +missing_from_mart as ( + select * from source_periods + except + select * from mart_periods +), + +unexpected_in_mart as ( + select * from mart_periods + except + select * from source_periods +) + +select * from missing_from_mart +union all +select * from unexpected_in_mart diff --git a/scripts/validate.sh b/scripts/validate.sh index 57bb5ae..e127d5d 100755 --- a/scripts/validate.sh +++ b/scripts/validate.sh @@ -2,6 +2,7 @@ set -euo pipefail uv run ruff check src tests +uv run ruff format --check src tests uv run mypy src uv run pytest uv run dbt parse --project-dir dbt --profiles-dir dbt diff --git a/src/ledger/config.py b/src/ledger/config.py index c103501..bcd9499 100644 --- a/src/ledger/config.py +++ b/src/ledger/config.py @@ -94,7 +94,5 @@ def load_config(path: str | Path, environ: Mapping[str, str] | None = None) -> L _env_value(environment, pipeline, "request_timeout_seconds_env", 30) ), max_retries=int(_env_value(environment, pipeline, "max_retries_env", 3)), - requests_per_second=float( - _env_value(environment, pipeline, "requests_per_second_env", 5) - ), + requests_per_second=float(_env_value(environment, pipeline, "requests_per_second_env", 5)), ) diff --git a/src/ledger/sec_client.py b/src/ledger/sec_client.py index 1d0879f..3af9f33 100644 --- a/src/ledger/sec_client.py +++ b/src/ledger/sec_client.py @@ -1,7 +1,9 @@ from __future__ import annotations +import gzip import json import time +import zlib from pathlib import Path from typing import Any, Protocol, cast from urllib.error import HTTPError, URLError @@ -19,6 +21,20 @@ def _decode_payload(content: str, source: str) -> dict[str, Any]: return cast(dict[str, Any], payload) +def _decompress_body(content: bytes, content_encoding: str | None, source: str) -> bytes: + encoding = (content_encoding or "").strip().lower() + if encoding in ("", "identity"): + return content + if encoding == "gzip": + return gzip.decompress(content) + if encoding == "deflate": + try: + return zlib.decompress(content) + except zlib.error: + return zlib.decompress(content, -zlib.MAX_WBITS) + raise ValueError(f"Unsupported Content-Encoding {encoding!r} from {source}") + + class SecCompanyFactsClient: BASE_URL = "https://data.sec.gov/api/xbrl/companyfacts/CIK{cik}.json" @@ -56,8 +72,21 @@ def fetch_companyfacts(self, cik: str) -> dict[str, Any]: try: self._last_request_at = time.monotonic() with urlopen(request, timeout=self.timeout_seconds) as response: # noqa: S310 - return _decode_payload(response.read().decode("utf-8"), url) - except (HTTPError, URLError, TimeoutError, json.JSONDecodeError) as error: + body = _decompress_body( + response.read(), response.headers.get("Content-Encoding"), url + ) + return _decode_payload(body.decode("utf-8"), url) + except ( + EOFError, + HTTPError, + URLError, + TimeoutError, + json.JSONDecodeError, + UnicodeDecodeError, + OSError, + ValueError, + zlib.error, + ) as error: last_error = error if attempt >= self.max_retries: break diff --git a/tests/fixtures/CIK0000320193.json b/tests/fixtures/CIK0000320193.json index f6a3583..43afd89 100644 --- a/tests/fixtures/CIK0000320193.json +++ b/tests/fixtures/CIK0000320193.json @@ -9,15 +9,26 @@ "units": { "USD": [ { - "val": 1000000, + "val": 250000, "accn": "0000000000-26-000001", "fy": 2026, - "fp": "FY", - "form": "10-K", - "filed": "2026-01-31", - "start": "2025-01-01", - "end": "2025-12-31", - "frame": "CY2025" + "fp": "Q3", + "form": "10-Q", + "filed": "2026-10-31", + "start": "2026-07-01", + "end": "2026-09-30", + "frame": "CY2026Q3" + }, + { + "val": 750000, + "accn": "0000000000-26-000001", + "fy": 2026, + "fp": "Q3", + "form": "10-Q", + "filed": "2026-10-31", + "start": "2026-01-01", + "end": "2026-09-30", + "frame": "CY2026Q3YTD" } ] } diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index cfc7eb4..289c186 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -8,14 +8,19 @@ FIXTURE_DIR = Path(__file__).parent / "fixtures" -def test_normalization_extracts_financial_observations() -> None: +def test_normalization_extracts_distinct_financial_periods() -> None: payload = json.loads((FIXTURE_DIR / "CIK0000320193.json").read_text(encoding="utf-8")) rows = normalize_companyfacts(payload) - assert len(rows) == 1 - assert rows[0]["cik"] == "0000320193" - assert rows[0]["fact_name"] == "RevenueFromContractWithCustomerExcludingAssessedTax" - assert rows[0]["value"] == 1000000 + assert len(rows) == 2 + assert {row["cik"] for row in rows} == {"0000320193"} + assert {row["fact_name"] for row in rows} == { + "RevenueFromContractWithCustomerExcludingAssessedTax" + } + assert {(row["period_start_date"], row["value"]) for row in rows} == { + ("2026-01-01", 750000), + ("2026-07-01", 250000), + } def test_pipeline_is_repeatable_and_writes_governed_outputs(tmp_path: Path) -> None: @@ -34,8 +39,8 @@ def test_pipeline_is_repeatable_and_writes_governed_outputs(tmp_path: Path) -> N second = run_pipeline(config, client) assert first.issuer_count == 2 - assert first.normalized_fact_count == 2 - assert second.normalized_fact_count == 2 + assert first.normalized_fact_count == 3 + assert second.normalized_fact_count == 3 rows = [ json.loads(line) for line in (tmp_path / "normalized" / "financial_facts.jsonl") diff --git a/tests/test_sec_client.py b/tests/test_sec_client.py new file mode 100644 index 0000000..50fb0fe --- /dev/null +++ b/tests/test_sec_client.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +import gzip +import json +import zlib +from collections.abc import Callable +from types import TracebackType + +import pytest + +from ledger.sec_client import SecCompanyFactsClient + + +class FakeResponse: + def __init__(self, body: bytes, content_encoding: str) -> None: + self.body = body + self.headers = {"Content-Encoding": content_encoding} + + def read(self) -> bytes: + return self.body + + def __enter__(self) -> FakeResponse: + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + return None + + +@pytest.mark.parametrize( + ("content_encoding", "compressor"), + [("gzip", gzip.compress), ("deflate", zlib.compress)], +) +def test_live_client_decodes_compressed_responses( + monkeypatch: pytest.MonkeyPatch, + content_encoding: str, + compressor: Callable[[bytes], bytes], +) -> None: + payload = {"cik": 320193, "entityName": "Example Device Company", "facts": {}} + response = FakeResponse( + compressor(json.dumps(payload).encode("utf-8")), + content_encoding, + ) + + def fake_urlopen(request: object, timeout: int) -> FakeResponse: + del request, timeout + return response + + monkeypatch.setattr("ledger.sec_client.urlopen", fake_urlopen) + client = SecCompanyFactsClient( + user_agent="Example Engineering data@example.org", + max_retries=0, + requests_per_second=1000, + ) + + assert client.fetch_companyfacts("320193") == payload + + +@pytest.mark.parametrize( + ("content_encoding", "compressor"), + [("gzip", gzip.compress), ("deflate", zlib.compress)], +) +def test_live_client_retries_truncated_compressed_responses( + monkeypatch: pytest.MonkeyPatch, + content_encoding: str, + compressor: Callable[[bytes], bytes], +) -> None: + payload = {"cik": 320193, "entityName": "Example Device Company", "facts": {}} + compressed = compressor(json.dumps(payload).encode("utf-8")) + responses = iter( + [ + FakeResponse(compressed[: len(compressed) // 2], content_encoding), + FakeResponse(compressed, content_encoding), + ] + ) + call_count = 0 + + def fake_urlopen(request: object, timeout: int) -> FakeResponse: + nonlocal call_count + del request, timeout + call_count += 1 + return next(responses) + + monkeypatch.setattr("ledger.sec_client.urlopen", fake_urlopen) + monkeypatch.setattr("ledger.sec_client.time.sleep", lambda _seconds: None) + client = SecCompanyFactsClient( + user_agent="Example Engineering data@example.org", + max_retries=1, + requests_per_second=1000, + ) + + assert client.fetch_companyfacts("320193") == payload + assert call_count == 2