Fix SEC decoding and financial fact grain

.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

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,

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

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

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)),
     )

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

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"
             }
           ]
         }

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")

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