| """SEC Company Facts loader for structured financial observations.""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| import re |
| from collections.abc import Mapping |
| from pathlib import Path |
|
|
| from graph.graph import EdgeRecord, GraphStore |
| from graph.real.corpus import RealGraph, SourceMetadata, relation_phrase |
|
|
|
|
| def _slug(value: str, *, limit: int = 120) -> str: |
| return re.sub(r"[^A-Za-z0-9]+", "_", value.strip()).strip("_")[:limit] or "unknown" |
|
|
|
|
| def load_sec_companyfacts( |
| path: str | Path, |
| source: SourceMetadata, |
| *, |
| max_observations: int | None = None, |
| ) -> RealGraph: |
| """Load one JSON object per company from the SEC Company Facts API.""" |
| nodes: dict[str, str] = {} |
| labels: dict[str, str] = {} |
| attributes: dict[str, dict[str, str]] = {} |
| edges: list[EdgeRecord] = [] |
| observations = 0 |
| with Path(path).open(encoding="utf-8") as handle: |
| for line in handle: |
| if not line.strip(): |
| continue |
| payload = json.loads(line) |
| cik = str(payload.get("cik") or payload.get("entity", {}).get("cik", "")) |
| entity_name = str( |
| payload.get("entityName") |
| or payload.get("entity", {}).get("name", "") |
| or cik |
| ) |
| if not cik: |
| continue |
| company_id = f"sec:company:{cik.zfill(10)}" |
| nodes[company_id] = "company" |
| labels[company_id] = entity_name |
| facts = payload.get("facts", {}) |
| if not isinstance(facts, Mapping): |
| continue |
| for taxonomy, concepts in facts.items(): |
| if not isinstance(concepts, Mapping): |
| continue |
| for concept, definition in concepts.items(): |
| concept_id = f"sec:concept:{taxonomy}:{concept}" |
| nodes[concept_id] = "financial_concept" |
| labels[concept_id] = str( |
| definition.get("label", concept) |
| if isinstance(definition, Mapping) |
| else concept |
| ) |
| units = ( |
| definition.get("units", {}) |
| if isinstance(definition, Mapping) |
| else {} |
| ) |
| if not isinstance(units, Mapping): |
| continue |
| for unit, rows in units.items(): |
| if not isinstance(rows, list): |
| continue |
| for row in rows: |
| if not isinstance(row, Mapping): |
| continue |
| observation_key = ":".join( |
| str(row.get(field, "")) |
| for field in ( |
| "fy", |
| "fp", |
| "form", |
| "filed", |
| "end", |
| "start", |
| "val", |
| ) |
| ) |
| observation_id = ( |
| f"sec:observation:{cik}:{taxonomy}:{concept}:" |
| f"{_slug(unit)}:{_slug(observation_key)}" |
| ) |
| nodes[observation_id] = "financial_observation" |
| attributes[observation_id] = { |
| "value": str(row.get("val", "")), |
| "unit": str(unit), |
| "period_start": str(row.get("start", "")), |
| "period_end": str(row.get("end", "")), |
| "fiscal_year": str(row.get("fy", "")), |
| "form": str(row.get("form", "")), |
| "filed": str(row.get("filed", "")), |
| "evidence_kind": "regulatory_filing_observation", |
| } |
| edges.extend( |
| ( |
| EdgeRecord( |
| company_id, |
| "reports_fact", |
| observation_id, |
| "company", |
| "financial_observation", |
| relation_phrase("reports_fact"), |
| { |
| "evidence_kind": "regulatory_filing_observation" |
| }, |
| ), |
| EdgeRecord( |
| observation_id, |
| "instance_of", |
| concept_id, |
| "financial_observation", |
| "financial_concept", |
| relation_phrase("instance_of"), |
| { |
| "evidence_kind": "regulatory_filing_observation" |
| }, |
| ), |
| ) |
| ) |
| period = str(row.get("end", "")) |
| if period: |
| period_id = f"sec:period:{period}" |
| nodes[period_id] = "reporting_period" |
| labels[period_id] = period |
| edges.append( |
| EdgeRecord( |
| observation_id, |
| "in_period", |
| period_id, |
| "financial_observation", |
| "reporting_period", |
| relation_phrase("in_period"), |
| { |
| "evidence_kind": "regulatory_filing_observation" |
| }, |
| ) |
| ) |
| observations += 1 |
| if ( |
| max_observations is not None |
| and observations >= max_observations |
| ): |
| return RealGraph( |
| graph=GraphStore( |
| nodes, |
| edges, |
| graph_version=f"{source.source_id}@{source.version}", |
| node_attributes=attributes, |
| relation_attributes={ |
| relation: {"functional": "false"} |
| for relation in ( |
| "reports_fact", |
| "instance_of", |
| "in_period", |
| ) |
| }, |
| ), |
| node_labels=labels, |
| relation_phrases={ |
| relation: relation_phrase(relation) |
| for relation in ( |
| "reports_fact", |
| "instance_of", |
| "in_period", |
| ) |
| }, |
| source=source, |
| ) |
| return RealGraph( |
| graph=GraphStore( |
| nodes, |
| edges, |
| graph_version=f"{source.source_id}@{source.version}", |
| node_attributes=attributes, |
| relation_attributes={ |
| relation: {"functional": "false"} |
| for relation in ("reports_fact", "instance_of", "in_period") |
| }, |
| ), |
| node_labels=labels, |
| relation_phrases={ |
| relation: relation_phrase(relation) |
| for relation in ("reports_fact", "instance_of", "in_period") |
| }, |
| source=source, |
| ) |
|
|
