mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-10-28 13:01:38 +00:00
869cfed125
* merge branch develop with main * bump desktop version * fix the merge * sync Cargo.lock to desktop 2.0.0 * fix cli conftest.py --------- Co-authored-by: deeleeramone <>
3559 lines
128 KiB
Python
3559 lines
128 KiB
Python
"""Unit tests for the statement_schema engine.
|
|
|
|
These tests drive the pure functions in
|
|
``openbb_sec.utils.statement_schema`` directly with small synthetic
|
|
inputs to exercise statement-type detection, schema lookups, and
|
|
row-level extraction. A real BlackRock company-facts fixture is also
|
|
pushed end-to-end (annual, quarterly, and growth periods) to cover the
|
|
combinatorial extraction/imputation branches that only realistic data
|
|
reaches. The imputation engine's own unit tests live in
|
|
``test_imputation.py``.
|
|
|
|
Tests only — no source under ``openbb_sec/`` is modified.
|
|
"""
|
|
|
|
# flake8: noqa: D101,D102,D103,D403
|
|
|
|
import json
|
|
from collections import Counter
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from openbb_sec.utils.company_facts import resolve_company_facts
|
|
from openbb_sec.utils.statement_schema import StatementSchema
|
|
from openbb_sec.utils.statement_schema._detection import (
|
|
_collapse_near_dates,
|
|
_fiscal_position,
|
|
_has_recent_data,
|
|
_latest_annual_shares,
|
|
annual_fiscal_years,
|
|
detect_reporting_currency,
|
|
detect_type,
|
|
get_filing_dates,
|
|
get_fiscal_meta,
|
|
preliminary_view,
|
|
prior_period_end,
|
|
)
|
|
from openbb_sec.utils.statement_schema._extraction import (
|
|
_get_annual_values,
|
|
_get_unit_data,
|
|
_get_ytd9_values,
|
|
compute_ref_filings,
|
|
extract_row_values,
|
|
rescaled_facts,
|
|
)
|
|
from openbb_sec.utils.statement_schema._types import (
|
|
SUPERSEDED_SUFFIX,
|
|
RowDef,
|
|
RowResult,
|
|
ScaledFacts,
|
|
_tolerance,
|
|
)
|
|
|
|
_FIXTURE_DIR = Path(__file__).parent / "record"
|
|
_M = 1_000_000 # magnitudes must exceed the 1M tolerance cap to register
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Builders
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _entry(
|
|
end, start=None, *, form="10-K", filed=None, fy=None, fp="FY", val=1.0, accn=None
|
|
):
|
|
"""Build a single XBRL fact entry dict; filed defaults to 45 days after end."""
|
|
if filed is None:
|
|
try:
|
|
filed = (datetime.strptime(end, "%Y-%m-%d") + timedelta(days=45)).strftime(
|
|
"%Y-%m-%d"
|
|
)
|
|
except ValueError:
|
|
filed = "2024-02-15"
|
|
e = {"end": end, "val": val, "form": form, "filed": filed, "fp": fp}
|
|
if accn is not None:
|
|
e["accn"] = accn
|
|
if start is not None:
|
|
e["start"] = start
|
|
if fy is not None:
|
|
e["fy"] = fy
|
|
elif end[:4].isdigit():
|
|
e["fy"] = int(end[:4])
|
|
else:
|
|
e["fy"] = 0
|
|
return e
|
|
|
|
|
|
_SHARES = "WeightedAverageNumberOfSharesOutstandingBasic"
|
|
|
|
|
|
def _cover(*counts):
|
|
"""Build dei cover-page share counts from (accession, filed, count) tuples."""
|
|
return (
|
|
"dei",
|
|
"EntityCommonStockSharesOutstanding",
|
|
"shares",
|
|
[_entry(filed, filed=filed, val=val, accn=accn) for accn, filed, val in counts],
|
|
)
|
|
|
|
|
|
def _facts(tags):
|
|
"""Build a facts dict: {ns: {tag: {units: {unit: [entries]}}}}.
|
|
|
|
``tags`` is an iterable of (ns, tag, unit, [entries]) tuples.
|
|
"""
|
|
out: dict = {}
|
|
for ns, tag, unit, entries in tags:
|
|
out.setdefault(ns, {}).setdefault(tag, {"units": {}})
|
|
out[ns][tag]["units"].setdefault(unit, []).extend(entries)
|
|
return out
|
|
|
|
|
|
def _rd(
|
|
tag,
|
|
*,
|
|
xbrl=(),
|
|
unit="monetary",
|
|
period_type="duration",
|
|
parent=None,
|
|
factor="+",
|
|
balance="",
|
|
sequence=1,
|
|
label=None,
|
|
):
|
|
"""Build a RowDef with a tuple of {tag, namespace} xbrl entries."""
|
|
return RowDef(
|
|
tag=tag,
|
|
label=label or tag.replace("_", " ").title(),
|
|
description="",
|
|
parent=parent,
|
|
sequence=sequence,
|
|
factor=factor,
|
|
balance=balance,
|
|
unit=unit,
|
|
period_type=period_type,
|
|
xbrl_tags=tuple({"tag": t, "namespace": ns} for t, ns in xbrl),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _types._tolerance
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTolerance:
|
|
def test_floor(self):
|
|
# Small magnitudes floor at 100k.
|
|
assert _tolerance(1.0, 2.0) == 100_000
|
|
|
|
def test_cap(self):
|
|
# 0.1% of 10B = 10M, capped at 1M.
|
|
assert _tolerance(10_000_000_000.0) == 1_000_000
|
|
|
|
def test_scaled_between(self):
|
|
# 0.1% of 500M = 500k, between floor and cap.
|
|
assert _tolerance(500_000_000.0) == 500_000
|
|
|
|
def test_ignores_none(self):
|
|
assert _tolerance(None, None) == 100_000
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _detection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
_DETECT_KW: dict[str, Any] = dict(
|
|
insurance_is_signals=["PremiumsEarnedNet", "BenefitsLossesAndExpenses"],
|
|
insurance_bs_signals=["LiabilityForFuturePolicyBenefits", "UnearnedPremiums"],
|
|
financial_signals=["InterestIncomeExpenseNet", "NoninterestIncome"],
|
|
min_financial_signals=2,
|
|
industrial_signals=["CostOfGoodsAndServicesSold", "CostOfRevenue"],
|
|
diversified_signals=["CostsAndExpenses", "OperatingIncomeLoss"],
|
|
)
|
|
|
|
|
|
class TestDetectType:
|
|
def test_insurance_wins_when_total_exceeds_financial(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"PremiumsEarnedNet",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01")],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"LiabilityForFuturePolicyBenefits",
|
|
"USD",
|
|
[_entry("2023-12-31")],
|
|
),
|
|
("us-gaap", "UnearnedPremiums", "USD", [_entry("2023-12-31")]),
|
|
(
|
|
"us-gaap",
|
|
"InterestIncomeExpenseNet",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01")],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"NoninterestIncome",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01")],
|
|
),
|
|
]
|
|
)
|
|
# ins_total = 3 (1 IS + 2 BS), fin_count = 2 -> insurance
|
|
assert detect_type(facts, **_DETECT_KW) == "insurance"
|
|
|
|
def test_financial_wins_when_count_exceeds_insurance_total(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"PremiumsEarnedNet",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01")],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"LiabilityForFuturePolicyBenefits",
|
|
"USD",
|
|
[_entry("2023-12-31")],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"InterestIncomeExpenseNet",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01")],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"NoninterestIncome",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01")],
|
|
),
|
|
]
|
|
)
|
|
# ins_total = 2, fin_count = 2 -> not > so financial branch:
|
|
# is_insurance and is_financial, ins_total(2) > fin_count(2) is False -> financial
|
|
assert detect_type(facts, **_DETECT_KW) == "financial"
|
|
|
|
def test_insurance_only(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"PremiumsEarnedNet",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01")],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"LiabilityForFuturePolicyBenefits",
|
|
"USD",
|
|
[_entry("2023-12-31")],
|
|
),
|
|
]
|
|
)
|
|
assert detect_type(facts, **_DETECT_KW) == "insurance"
|
|
|
|
def test_financial_only(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"InterestIncomeExpenseNet",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01")],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"NoninterestIncome",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01")],
|
|
),
|
|
]
|
|
)
|
|
assert detect_type(facts, **_DETECT_KW) == "financial"
|
|
|
|
def test_industrial_requires_recent_data(self):
|
|
# COGS present and recent (within 5 years of an annual filing) -> industrial.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"CostOfGoodsAndServicesSold",
|
|
"USD",
|
|
[_entry("2024-12-31", "2024-01-01", form="10-K")],
|
|
),
|
|
]
|
|
)
|
|
assert detect_type(facts, **_DETECT_KW) == "industrial"
|
|
|
|
def test_stale_cogs_falls_through_to_diversified(self):
|
|
# COGS only present in an ancient filing -> not "recent" -> diversified via CostsAndExpenses.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"CostOfGoodsAndServicesSold",
|
|
"USD",
|
|
[_entry("2005-12-31", "2005-01-01", form="10-K")],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"CostsAndExpenses",
|
|
"USD",
|
|
[_entry("2024-12-31", "2024-01-01")],
|
|
),
|
|
]
|
|
)
|
|
assert detect_type(facts, **_DETECT_KW) == "diversified"
|
|
|
|
def test_default_industrial_when_no_signals(self):
|
|
facts = _facts([("us-gaap", "SomeRandomTag", "USD", [_entry("2023-12-31")])])
|
|
assert detect_type(facts, **_DETECT_KW) == "industrial"
|
|
|
|
def test_non_dict_namespace_skipped(self):
|
|
# detect_type iterates facts.values(); non-dict values are ignored.
|
|
facts = {
|
|
"weird": ["not", "a", "dict"],
|
|
"us-gaap": {"CostsAndExpenses": {"units": {}}},
|
|
}
|
|
assert detect_type(facts, **_DETECT_KW) == "diversified"
|
|
|
|
|
|
class TestLatestAnnualShares:
|
|
def test_null_revenue_value_skipped(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry("2023-12-31", "2023-01-01", val=None, accn="a1"),
|
|
_entry("2023-12-31", "2023-01-01", val=200.0, accn="a1"),
|
|
],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"PremiumsEarnedNet",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=50.0, accn="a1")],
|
|
),
|
|
]
|
|
)
|
|
revenue, shares = _latest_annual_shares(
|
|
facts, ["Revenues"], {"insurance": ["PremiumsEarnedNet"]}
|
|
)
|
|
assert revenue == 200.0
|
|
assert shares == {"insurance": 0.25}
|
|
|
|
def test_zero_revenue_returns_no_shares(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=0.0, accn="a1")],
|
|
)
|
|
]
|
|
)
|
|
assert _latest_annual_shares(
|
|
facts, ["Revenues"], {"insurance": ["PremiumsEarnedNet"]}
|
|
) == (None, {"insurance": None})
|
|
|
|
def test_non_dict_namespace_skipped_in_measure_lookup(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=100.0)],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"InterestIncomeOperating",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=40.0)],
|
|
),
|
|
]
|
|
)
|
|
facts["weird"] = ["not", "a", "dict"]
|
|
revenue, shares = _latest_annual_shares(
|
|
facts, ["Revenues"], {"financial": ["InterestIncomeOperating"]}
|
|
)
|
|
assert revenue == 100.0
|
|
assert shares == {"financial": 0.4}
|
|
|
|
|
|
class TestCollapseNearDates:
|
|
def test_interim_near_fiscal_year_end_dropped(self):
|
|
kept = _collapse_near_dates(
|
|
{"2023-03-31", "2023-12-28"}, {"2023-12-31"}, Counter()
|
|
)
|
|
assert kept == {"2023-03-31"}
|
|
|
|
def test_better_supported_interim_kept(self):
|
|
weights = Counter({"2023-06-30": 1, "2023-07-02": 3})
|
|
assert _collapse_near_dates({"2023-06-30", "2023-07-02"}, set(), weights) == {
|
|
"2023-07-02"
|
|
}
|
|
|
|
def test_first_interim_kept_when_not_outweighed(self):
|
|
weights = Counter({"2023-06-30": 3, "2023-07-02": 1})
|
|
assert _collapse_near_dates({"2023-06-30", "2023-07-02"}, set(), weights) == {
|
|
"2023-06-30"
|
|
}
|
|
|
|
|
|
class TestMalformedDates:
|
|
def test_filing_dates_skip_unparseable_end(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-02-30", "2022-03-01", filed="2023-05-01")],
|
|
)
|
|
]
|
|
)
|
|
assert get_filing_dates(facts, "annual") == set()
|
|
|
|
def test_annual_fiscal_years_skip_unparseable_end(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-02-30", "2022-03-01", filed="2023-05-01")],
|
|
)
|
|
]
|
|
)
|
|
assert annual_fiscal_years(facts, set()) == {}
|
|
|
|
|
|
class TestFiscalPosition:
|
|
_ENDS = ["2022-12-31", "2023-12-31"]
|
|
_YEARS = {"2022-12-31": 2022, "2023-12-31": 2023}
|
|
|
|
def test_date_before_grid_rolls_back(self):
|
|
assert _fiscal_position("2020-06-30", self._ENDS, self._YEARS) == (2020, 2)
|
|
|
|
def test_date_after_grid_rolls_forward(self):
|
|
assert _fiscal_position("2025-06-30", self._ENDS, self._YEARS) == (2025, 2)
|
|
|
|
|
|
class TestPreliminaryView:
|
|
def _facts(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry("2023-12-31", "2023-01-01", filed="2024-02-15"),
|
|
_entry(
|
|
"2023-12-31", "2023-01-01", form="8-K", filed="2024-01-25"
|
|
),
|
|
_entry(
|
|
"2024-12-31", "2024-01-01", form="8-K", filed="2025-01-25"
|
|
),
|
|
],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"Assets",
|
|
"USD",
|
|
[_entry("2023-12-31", filed="2024-02-15")],
|
|
),
|
|
(
|
|
"dei",
|
|
"EntityCommonStockSharesOutstanding",
|
|
"shares",
|
|
[_entry("2024-01-31", form="10-K", filed="2024-02-15")],
|
|
),
|
|
]
|
|
)
|
|
facts["weird"] = ["not", "a", "dict"]
|
|
return facts
|
|
|
|
def test_reported_period_superseded_and_unreported_kept(self):
|
|
facts = self._facts()
|
|
view = preliminary_view(facts)
|
|
forms = [e["form"] for e in view["us-gaap"]["Revenues"]["units"]["USD"]]
|
|
assert forms == ["10-K", "8-K" + SUPERSEDED_SUFFIX, "8-K"]
|
|
assert view["us-gaap"]["Assets"] is facts["us-gaap"]["Assets"]
|
|
assert view["weird"] is facts["weird"]
|
|
assert view["dei"] == facts["dei"]
|
|
|
|
def test_pit_mode_keeps_entry_filed_before_regular_filing(self):
|
|
view = preliminary_view(self._facts(), pit_mode=True)
|
|
forms = [e["form"] for e in view["us-gaap"]["Revenues"]["units"]["USD"]]
|
|
assert forms == ["10-K", "8-K", "8-K"]
|
|
|
|
|
|
class TestHasRecentData:
|
|
def test_true_for_recent_annual(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"CostOfRevenue",
|
|
"USD",
|
|
[_entry("2024-12-31", "2024-01-01", form="10-K")],
|
|
)
|
|
]
|
|
)
|
|
assert _has_recent_data(facts, "CostOfRevenue") is True
|
|
|
|
def test_false_for_old(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"CostOfRevenue",
|
|
"USD",
|
|
[_entry("2005-12-31", "2005-01-01", form="10-K")],
|
|
)
|
|
]
|
|
)
|
|
assert _has_recent_data(facts, "CostOfRevenue") is False
|
|
|
|
def test_false_for_non_annual_form(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"CostOfRevenue",
|
|
"USD",
|
|
[_entry("2024-09-30", "2024-07-01", form="10-Q")],
|
|
)
|
|
]
|
|
)
|
|
assert _has_recent_data(facts, "CostOfRevenue") is False
|
|
|
|
def test_missing_tag(self):
|
|
facts = _facts([("us-gaap", "Other", "USD", [_entry("2024-12-31")])])
|
|
assert _has_recent_data(facts, "CostOfRevenue") is False
|
|
|
|
|
|
class TestGetFilingDates:
|
|
def test_annual_basic(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Assets",
|
|
"USD",
|
|
[_entry(f"{y}-12-31") for y in (2022, 2023)],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry(f"{y}-12-31", f"{y}-01-01") for y in (2022, 2023)],
|
|
),
|
|
]
|
|
)
|
|
assert get_filing_dates(facts, "annual") == {"2022-12-31", "2023-12-31"}
|
|
|
|
def test_bad_dates_skipped(self):
|
|
# start==end and unparseable dates are ignored, leaving no annual dates.
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "Revenues", "USD", [_entry("2023-12-31", "2023-12-31")]),
|
|
("us-gaap", "Other", "USD", [_entry("bad", "worse")]),
|
|
]
|
|
)
|
|
assert get_filing_dates(facts, "annual") == set()
|
|
|
|
def test_quarterly_merges_annual_and_drops_orphan_quarter(self):
|
|
# A quarter end with no instant Assets anchor and earlier than others gets dropped.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Assets",
|
|
"USD",
|
|
[
|
|
_entry("2023-03-31", form="10-Q"),
|
|
_entry("2023-06-30", form="10-Q"),
|
|
_entry("2023-12-31", form="10-K"),
|
|
],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry("2023-03-31", "2023-01-01", form="10-Q"),
|
|
_entry("2023-06-30", "2023-04-01", form="10-Q"),
|
|
_entry("2023-12-31", "2023-01-01", form="10-K"),
|
|
],
|
|
),
|
|
]
|
|
)
|
|
dates = get_filing_dates(facts, "quarterly")
|
|
assert "2023-03-31" in dates
|
|
assert "2023-12-31" in dates # annual merged in
|
|
|
|
def test_canonical_month_filter_keeps_offcycle_without_neighbor(self):
|
|
# >3 annual dates, dominant 12-31, plus an isolated 06-30 far from any canonical.
|
|
rev = [
|
|
_entry(f"{y}-12-31", f"{y}-01-01") for y in (2020, 2021, 2022, 2023, 2024)
|
|
]
|
|
assets = [_entry(f"{y}-12-31") for y in (2020, 2021, 2022, 2023, 2024)]
|
|
rev.append(_entry("2016-06-30", "2015-07-01"))
|
|
assets.append(_entry("2016-06-30"))
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "Assets", "USD", assets),
|
|
("us-gaap", "Revenues", "USD", rev),
|
|
]
|
|
)
|
|
dates = get_filing_dates(facts, "annual")
|
|
assert "2016-06-30" in dates
|
|
assert "2024-12-31" in dates
|
|
|
|
def test_canonical_month_filter_drops_near_duplicate(self):
|
|
# Two canonical dates within 10 days: the non-exact one is discarded.
|
|
rev = [_entry(f"{y}-12-31", f"{y}-01-01") for y in (2021, 2022, 2023, 2024)]
|
|
assets = [_entry(f"{y}-12-31") for y in (2021, 2022, 2023, 2024)]
|
|
# Near-duplicate 52-week fiscal end a few days before a calendar 12-31.
|
|
rev.append(_entry("2024-12-28", "2024-01-01"))
|
|
assets.append(_entry("2024-12-28"))
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "Assets", "USD", assets),
|
|
("us-gaap", "Revenues", "USD", rev),
|
|
]
|
|
)
|
|
dates = get_filing_dates(facts, "annual")
|
|
# The exact 12-31 wins over the 12-28 near-duplicate.
|
|
assert "2024-12-31" in dates
|
|
assert "2024-12-28" not in dates
|
|
|
|
def test_include_preliminary_adds_8k(self):
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "Assets", "USD", [_entry("2024-12-31")]),
|
|
("us-gaap", "Revenues", "USD", [_entry("2024-12-31", "2024-01-01")]),
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2025-12-31", "2025-01-01", form="8-K")],
|
|
),
|
|
]
|
|
)
|
|
with_prelim = get_filing_dates(facts, "annual", include_preliminary=True)
|
|
assert "2025-12-31" in with_prelim
|
|
without = get_filing_dates(facts, "annual")
|
|
assert "2025-12-31" not in without
|
|
|
|
|
|
class TestGetFiscalMeta:
|
|
def test_annual_descending_year_correction(self):
|
|
# Two annual dates where the earlier filing erroneously claims a >= fy;
|
|
# get_fiscal_meta forces the earlier to be next_fy - 1.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Assets",
|
|
"USD",
|
|
[_entry("2022-12-31"), _entry("2023-12-31")],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry("2022-12-31", "2022-01-01", fy=2023, fp="FY"),
|
|
_entry("2023-12-31", "2023-01-01", fy=2023, fp="FY"),
|
|
],
|
|
),
|
|
]
|
|
)
|
|
dates = get_filing_dates(facts, "annual")
|
|
meta = get_fiscal_meta(facts, "annual", dates)
|
|
assert meta["2023-12-31"]["fiscal_year"] == 2023
|
|
assert meta["2022-12-31"]["fiscal_year"] == 2022 # corrected down
|
|
|
|
def test_annual_fallback_year_from_date(self):
|
|
# No fy/fp metadata at all -> falls back to int(date[:4]).
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "Assets", "USD", [_entry("2023-12-31", fy=None, fp="")]),
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", fp="")],
|
|
),
|
|
]
|
|
)
|
|
dates = {"2023-12-31"}
|
|
meta = get_fiscal_meta(facts, "annual", dates)
|
|
assert meta["2023-12-31"] == {"fiscal_year": 2023, "fiscal_period": "FY"}
|
|
|
|
def test_quarterly_period_from_10q(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Assets",
|
|
"USD",
|
|
[
|
|
_entry("2023-09-30", form="10-Q", fy=2023, fp="Q3"),
|
|
_entry("2023-12-31", form="10-K", fy=2023, fp="FY"),
|
|
],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-09-30", "2023-07-01", form="10-Q", fy=2023, fp="Q3"
|
|
),
|
|
_entry(
|
|
"2023-12-31", "2023-01-01", form="10-K", fy=2023, fp="FY"
|
|
),
|
|
],
|
|
),
|
|
]
|
|
)
|
|
dates = get_filing_dates(facts, "quarterly")
|
|
meta = get_fiscal_meta(facts, "quarterly", dates)
|
|
assert meta["2023-09-30"]["fiscal_period"] == "Q3"
|
|
# Annual date present in quarterly extraction becomes Q4.
|
|
assert meta["2023-12-31"]["fiscal_period"] == "Q4"
|
|
|
|
|
|
class TestDetectCurrency:
|
|
def test_picks_most_common(self):
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "A", "EUR", [_entry("2023-12-31")]),
|
|
("us-gaap", "B", "EUR", [_entry("2023-12-31")]),
|
|
("us-gaap", "C", "USD", [_entry("2023-12-31")]),
|
|
]
|
|
)
|
|
assert detect_reporting_currency(facts) == "EUR"
|
|
|
|
def test_skips_shares_and_ratios(self):
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "Shares", "shares", [_entry("2023-12-31")]),
|
|
("us-gaap", "Ratio", "pure", [_entry("2023-12-31")]),
|
|
("us-gaap", "PerShare", "USD/shares", [_entry("2023-12-31")]),
|
|
]
|
|
)
|
|
# Only currency-looking 3-letter upper keys count; none here -> USD default.
|
|
assert detect_reporting_currency(facts) == "USD"
|
|
|
|
def test_empty_defaults_usd(self):
|
|
assert detect_reporting_currency({}) == "USD"
|
|
|
|
|
|
class TestPriorPeriodEnd:
|
|
@pytest.mark.parametrize(
|
|
"date,expected",
|
|
[
|
|
("2023-03-31", "2022-12-31"),
|
|
("2023-06-30", "2023-03-31"),
|
|
("2023-09-30", "2023-06-30"),
|
|
("2023-12-31", "2023-09-30"),
|
|
("2023-01-31", "2022-10-31"),
|
|
("2023-04-30", "2023-01-31"),
|
|
],
|
|
)
|
|
def test_known_months(self, date, expected):
|
|
assert prior_period_end(date) == expected
|
|
|
|
def test_unmapped_month_returns_none(self):
|
|
assert prior_period_end("2023-05-31") is None
|
|
|
|
def test_bad_date_returns_none(self):
|
|
assert prior_period_end("garbage") is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _extraction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetUnitData:
|
|
def test_shares_key(self):
|
|
td = {"units": {"shares": [{"val": 1}], "USD": [{"val": 2}]}}
|
|
assert _get_unit_data(td, "shares") == [{"val": 1}]
|
|
|
|
def test_per_share_exact(self):
|
|
td = {"units": {"USD/shares": [{"val": 1.5}]}}
|
|
assert _get_unit_data(td, "per_share", "USD") == [{"val": 1.5}]
|
|
|
|
def test_per_share_prefix_fallback(self):
|
|
td = {"units": {"USD/foo": [{"val": 9}]}}
|
|
assert _get_unit_data(td, "per_share", "USD") == [{"val": 9}]
|
|
|
|
def test_monetary_currency_match(self):
|
|
td = {"units": {"EUR": [{"val": 3}]}}
|
|
assert _get_unit_data(td, "monetary", "EUR") == [{"val": 3}]
|
|
|
|
def test_falls_back_to_first_unit(self):
|
|
td = {"units": {"GBP": [{"val": 7}]}}
|
|
# Requesting USD monetary but only GBP present -> first unit returned.
|
|
assert _get_unit_data(td, "monetary", "USD") == [{"val": 7}]
|
|
|
|
def test_no_units_returns_none(self):
|
|
assert _get_unit_data({"units": {}}, "monetary") is None
|
|
|
|
|
|
class TestGetAnnualValues:
|
|
def test_instant_returns_empty(self):
|
|
row = _rd("x", period_type="instant", xbrl=[("Assets", "us-gaap")])
|
|
assert _get_annual_values({}, row) == {}
|
|
|
|
def test_picks_earliest_filed(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-12-31", "2023-01-01", filed="2024-02-01", val=500.0
|
|
),
|
|
_entry(
|
|
"2023-12-31", "2023-01-01", filed="2025-02-01", val=510.0
|
|
),
|
|
],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
res = _get_annual_values(facts, row)
|
|
assert res["2023-12-31"][1] == 500.0 # earliest filing wins
|
|
assert res["2023-12-31"][2] == "us-gaap:Revenues"
|
|
|
|
def test_tag_chain_priority(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"RevA",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=100.0)],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"RevB",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=200.0)],
|
|
),
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("RevA", "us-gaap"), ("RevB", "us-gaap")])
|
|
res = _get_annual_values(facts, row)
|
|
assert res["2023-12-31"][1] == 100.0 # first tag in chain
|
|
|
|
def test_duration_window_filter(self):
|
|
# A 200-day duration is not annual (300-400) -> excluded.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-07-01", "2022-12-01", val=9.0)],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert _get_annual_values(facts, row) == {}
|
|
|
|
def test_reference_map_selects_mapped_filing(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-12-31", "2023-01-01", filed="2024-02-01", val=500.0
|
|
),
|
|
_entry(
|
|
"2023-12-31", "2023-01-01", filed="2025-02-01", val=510.0
|
|
),
|
|
],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
res = _get_annual_values(facts, row, ref_filed_map={"2023-12-31": "2025-02-01"})
|
|
assert res["2023-12-31"] == ("2023-01-01", 510.0, "us-gaap:Revenues")
|
|
|
|
def test_reference_map_falls_back_to_latest_earlier_filing(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-12-31", "2023-01-01", filed="2024-02-01", val=500.0
|
|
),
|
|
_entry(
|
|
"2023-12-31", "2023-01-01", filed="2024-05-01", val=505.0
|
|
),
|
|
],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
res = _get_annual_values(facts, row, ref_filed_map={"2023-12-31": "2025-02-01"})
|
|
assert res["2023-12-31"] == (
|
|
"2023-01-01",
|
|
505.0,
|
|
"us-gaap:Revenues(fallback)",
|
|
)
|
|
|
|
def test_reference_map_falls_back_to_earliest_later_filing(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-12-31", "2023-01-01", filed="2025-02-01", val=510.0
|
|
),
|
|
_entry(
|
|
"2023-12-31", "2023-01-01", filed="2025-05-01", val=515.0
|
|
),
|
|
],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
res = _get_annual_values(facts, row, ref_filed_map={"2023-12-31": "2024-02-01"})
|
|
assert res["2023-12-31"][1:] == (510.0, "us-gaap:Revenues(fallback)")
|
|
|
|
def test_reference_map_skips_unmapped_period(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=500.0)],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert _get_annual_values(facts, row, ref_filed_map={}) == {}
|
|
|
|
|
|
class TestGetYtd9Values:
|
|
def test_instant_empty(self):
|
|
row = _rd("x", period_type="instant", xbrl=[("A", "us-gaap")])
|
|
assert _get_ytd9_values({}, row) == {}
|
|
|
|
def test_q4_from_fy_minus_ytd9(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-12-31", "2023-01-01", filed="2024-02-01", val=1000.0
|
|
),
|
|
_entry(
|
|
"2023-09-30", "2023-01-01", filed="2024-02-01", val=700.0
|
|
),
|
|
],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
res = _get_ytd9_values(facts, row)
|
|
assert res["2023-12-31"] == 300.0 # 1000 - 700
|
|
|
|
def test_missing_ytd_returns_empty(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=1000.0)],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert _get_ytd9_values(facts, row) == {}
|
|
|
|
|
|
class TestExtractRowValues:
|
|
def test_annual_simple(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=500.0)],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
vals, srcs = extract_row_values(facts, row, "annual", "USD")
|
|
assert vals["2023-12-31"] == 500.0
|
|
assert srcs["2023-12-31"] == "us-gaap:Revenues"
|
|
|
|
def test_instant_balance_sheet(self):
|
|
facts = _facts(
|
|
[("us-gaap", "Assets", "USD", [_entry("2023-12-31", val=999.0)])]
|
|
)
|
|
row = _rd("total_assets", period_type="instant", xbrl=[("Assets", "us-gaap")])
|
|
vals, _ = extract_row_values(facts, row, "annual", "USD")
|
|
assert vals["2023-12-31"] == 999.0
|
|
|
|
def test_cross_target_identity_lock(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"ProfitLoss",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=42.0)],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("net_income", xbrl=[("ProfitLoss", "us-gaap")])
|
|
vals, srcs = extract_row_values(
|
|
facts,
|
|
row,
|
|
"annual",
|
|
"USD",
|
|
cross_targets={"2023-12-31": 42.0},
|
|
statement="cash_flow",
|
|
)
|
|
assert vals["2023-12-31"] == 42.0
|
|
assert "identity_lock:cash_flow" in srcs["2023-12-31"]
|
|
|
|
def test_identity_locked_quarter_keeps_its_concept_for_ytd(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"A",
|
|
"USD",
|
|
[
|
|
_entry("2023-03-31", "2023-01-01", form="10-Q", val=100.0),
|
|
_entry("2023-06-30", "2023-01-01", form="10-Q", val=200.0),
|
|
],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"B",
|
|
"USD",
|
|
[
|
|
_entry("2023-03-31", "2023-01-01", form="10-Q", val=205.0),
|
|
_entry("2023-06-30", "2023-04-01", form="10-Q", val=207.0),
|
|
],
|
|
),
|
|
]
|
|
)
|
|
row = _rd(
|
|
"depreciation_and_amortization",
|
|
xbrl=[("A", "us-gaap"), ("B", "us-gaap")],
|
|
)
|
|
vals, srcs = extract_row_values(
|
|
facts,
|
|
row,
|
|
"quarterly",
|
|
"USD",
|
|
cross_targets={"2023-03-31": 100.0},
|
|
statement="cash_flow",
|
|
)
|
|
assert vals["2023-06-30"] == 100.0
|
|
assert srcs["2023-06-30"] == "ytd_derived(us-gaap:A)"
|
|
|
|
def test_share_count_scale_restated_before_extraction(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
_SHARES,
|
|
"shares",
|
|
[
|
|
_entry(
|
|
"2017-12-31",
|
|
"2017-01-01",
|
|
filed="2019-02-21",
|
|
val=351.9,
|
|
accn="x",
|
|
),
|
|
],
|
|
),
|
|
_cover(("x", "2019-02-21", 350_000_000), ("y", "2018-05-01", 352 * _M)),
|
|
]
|
|
)
|
|
row = _rd(
|
|
"weighted_ave_basic_shares_os", unit="shares", xbrl=[(_SHARES, "us-gaap")]
|
|
)
|
|
vals, srcs = extract_row_values(
|
|
rescaled_facts(facts),
|
|
row,
|
|
"annual",
|
|
"USD",
|
|
ref_filed_map={"2017-12-31": "2019-02-21"},
|
|
)
|
|
assert vals["2017-12-31"] == 351_900_000
|
|
assert isinstance(vals["2017-12-31"], int)
|
|
assert srcs["2017-12-31"] == f"us-gaap:{_SHARES}"
|
|
|
|
def test_ref_filed_fallback_when_exact_missing(self):
|
|
# ref_filed_map points to a date not in the filings -> fallback path.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", filed="2024-02-01", val=500.0)],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
vals, srcs = extract_row_values(
|
|
facts, row, "annual", "USD", ref_filed_map={"2023-12-31": "2099-01-01"}
|
|
)
|
|
assert vals["2023-12-31"] == 500.0
|
|
assert "(fallback)" in srcs["2023-12-31"]
|
|
|
|
def test_quarterly_q4_derivation_from_fy_minus_quarters(self):
|
|
# Three 10-Q quarters + FY -> Q4 = FY - sum(Q1..Q3).
|
|
entries = [
|
|
_entry("2023-03-31", "2023-01-01", form="10-Q", val=100.0),
|
|
_entry("2023-06-30", "2023-04-01", form="10-Q", val=150.0),
|
|
_entry("2023-09-30", "2023-07-01", form="10-Q", val=200.0),
|
|
_entry("2023-12-31", "2023-01-01", form="10-K", val=600.0),
|
|
]
|
|
facts = _facts([("us-gaap", "Revenues", "USD", entries)])
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
vals, srcs = extract_row_values(facts, row, "quarterly", "USD")
|
|
assert vals["2023-12-31"] == 150.0 # 600 - (100+150+200)
|
|
assert "Q4:" in srcs["2023-12-31"]
|
|
|
|
def test_quarterly_q4_uses_annual_reference_map(self):
|
|
entries = [
|
|
_entry("2023-03-31", "2023-01-01", form="10-Q", val=100.0),
|
|
_entry("2023-06-30", "2023-04-01", form="10-Q", val=150.0),
|
|
_entry("2023-09-30", "2023-07-01", form="10-Q", val=200.0),
|
|
_entry("2023-12-31", "2023-01-01", filed="2024-02-15", val=600.0),
|
|
_entry("2023-12-31", "2023-01-01", filed="2025-02-15", val=660.0),
|
|
]
|
|
facts = _facts([("us-gaap", "Revenues", "USD", entries)])
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
vals, srcs = extract_row_values(
|
|
facts,
|
|
row,
|
|
"quarterly",
|
|
"USD",
|
|
annual_ref_map={"2023-12-31": "2025-02-15"},
|
|
)
|
|
assert vals["2023-12-31"] == 210.0 # 660 - (100+150+200)
|
|
assert srcs["2023-12-31"].startswith("Q4: FY[us-gaap:Revenues]")
|
|
|
|
def test_quarterly_h2_derivation_single_interim(self):
|
|
# Semi-annual reporter: FY + one H1 interim -> H2 = FY - H1.
|
|
entries = [
|
|
_entry("2023-06-30", "2023-01-01", form="6-K", val=400.0),
|
|
_entry("2023-12-31", "2023-01-01", form="10-K", val=900.0),
|
|
]
|
|
facts = _facts([("us-gaap", "Revenues", "USD", entries)])
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
vals, srcs = extract_row_values(facts, row, "quarterly", "USD")
|
|
assert vals["2023-12-31"] == 500.0 # 900 - 400
|
|
assert "H2:" in srcs["2023-12-31"]
|
|
|
|
def test_shares_unit_has_no_q4_derivation(self):
|
|
entries = [_entry("2023-12-31", "2023-01-01", form="10-K", val=1_000.0)]
|
|
facts = _facts([("us-gaap", "WAS", "shares", entries)])
|
|
row = _rd(
|
|
"weighted_average_shares_outstanding",
|
|
unit="shares",
|
|
xbrl=[("WAS", "us-gaap")],
|
|
)
|
|
vals, srcs = extract_row_values(facts, row, "quarterly", "USD")
|
|
assert "2023-12-31" not in vals
|
|
assert "2023-12-31" not in srcs
|
|
|
|
def test_missing_tag_returns_empty(self):
|
|
row = _rd("total_revenue", xbrl=[("Nope", "us-gaap")])
|
|
vals, srcs = extract_row_values({}, row, "annual", "USD")
|
|
assert vals == {} and srcs == {}
|
|
|
|
def test_quarterly_ignores_annual_form_interim_monetary_facts(self):
|
|
entries = [
|
|
_entry("2023-03-31", "2023-01-01", form="10-K", val=100.0),
|
|
_entry("2023-12-31", "2023-01-01", form="10-K", val=600.0),
|
|
]
|
|
facts = _facts([("us-gaap", "Revenues", "USD", entries)])
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
vals, _ = extract_row_values(facts, row, "quarterly", "USD")
|
|
assert "2023-03-31" not in vals
|
|
|
|
def test_quarterly_keeps_annual_form_interim_share_facts(self):
|
|
entries = [_entry("2023-12-31", "2023-10-01", form="10-K", val=7.0)]
|
|
facts = _facts([("us-gaap", "WAS", "shares", entries)])
|
|
row = _rd("shares", unit="shares", xbrl=[("WAS", "us-gaap")])
|
|
vals, _ = extract_row_values(facts, row, "quarterly", "USD")
|
|
assert vals["2023-12-31"] == 7.0
|
|
|
|
def test_quarterly_q4_from_quarters_concept(self):
|
|
quarters = [
|
|
_entry("2023-03-31", "2023-01-01", form="10-Q", val=200.0),
|
|
_entry("2023-06-30", "2023-04-01", form="10-Q", val=250.0),
|
|
_entry("2023-09-30", "2023-07-01", form="10-Q", val=300.0),
|
|
_entry("2023-12-31", "2023-01-01", form="10-K", val=1000.0),
|
|
]
|
|
subset = [_entry("2023-12-31", "2023-01-01", form="10-K", val=700.0)]
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "Revenues", "USD", quarters),
|
|
(
|
|
"us-gaap",
|
|
"RevenueFromContractWithCustomerExcludingAssessedTax",
|
|
"USD",
|
|
subset,
|
|
),
|
|
]
|
|
)
|
|
row = _rd(
|
|
"operating_revenue",
|
|
xbrl=[
|
|
("RevenueFromContractWithCustomerExcludingAssessedTax", "us-gaap"),
|
|
("Revenues", "us-gaap"),
|
|
],
|
|
)
|
|
vals, srcs = extract_row_values(facts, row, "quarterly", "USD")
|
|
assert vals["2023-12-31"] == 250.0
|
|
assert srcs["2023-12-31"].startswith("Q4: FY[us-gaap:Revenues]")
|
|
|
|
def test_quarterly_q4_from_annual_concept_without_quarters_concept(self):
|
|
entries = [
|
|
_entry("2023-03-31", "2023-01-01", form="10-Q", val=200.0),
|
|
_entry("2023-06-30", "2023-04-01", form="10-Q", val=250.0),
|
|
_entry("2023-09-30", "2023-07-01", form="10-Q", val=300.0),
|
|
]
|
|
subset = [_entry("2023-12-31", "2023-01-01", form="10-K", val=700.0)]
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "Revenues", "USD", entries),
|
|
(
|
|
"us-gaap",
|
|
"RevenueFromContractWithCustomerExcludingAssessedTax",
|
|
"USD",
|
|
subset,
|
|
),
|
|
]
|
|
)
|
|
row = _rd(
|
|
"operating_revenue",
|
|
xbrl=[
|
|
("RevenueFromContractWithCustomerExcludingAssessedTax", "us-gaap"),
|
|
("Revenues", "us-gaap"),
|
|
],
|
|
)
|
|
vals, srcs = extract_row_values(facts, row, "quarterly", "USD")
|
|
assert vals["2023-12-31"] == -50.0
|
|
assert srcs["2023-12-31"].startswith(
|
|
"Q4: FY[us-gaap:RevenueFromContractWithCustomerExcludingAssessedTax]"
|
|
)
|
|
|
|
def test_instant_row_skips_duration_facts(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Assets",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=5.0)],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_assets", period_type="instant", xbrl=[("Assets", "us-gaap")])
|
|
assert extract_row_values(facts, row, "annual", "USD") == ({}, {})
|
|
|
|
def test_negative_net_position_skipped_in_asset_row(self):
|
|
net = "DeferredTaxAssetsLiabilitiesNetNoncurrent"
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
net,
|
|
"USD",
|
|
[_entry("2022-12-31", val=40.0), _entry("2023-12-31", val=-50.0)],
|
|
)
|
|
]
|
|
)
|
|
row = _rd(
|
|
"noncurrent_deferred_tax_assets",
|
|
period_type="instant",
|
|
balance="debit",
|
|
xbrl=[(net, "us-gaap")],
|
|
)
|
|
vals, _ = extract_row_values(facts, row, "annual", "USD")
|
|
assert vals == {"2022-12-31": 40.0}
|
|
|
|
|
|
class TestRescaledFacts:
|
|
def test_filing_scaled_in_full(self):
|
|
entries = [
|
|
_entry(f"{y}-12-31", f"{y}-01-01", filed=filed, val=val, accn=accn)
|
|
for y in range(2001, 2011)
|
|
for accn, filed, val in (
|
|
("bad", "2012-02-01", y),
|
|
("b", "2012-03-01", y * _M),
|
|
("c", "2013-03-01", y * _M),
|
|
)
|
|
]
|
|
entries.append(
|
|
_entry("2011-12-31", "2011-01-01", filed="2012-02-01", val=2.5, accn="bad")
|
|
)
|
|
facts = _facts([("us-gaap", "Revenues", "USD", entries)])
|
|
out = rescaled_facts(facts)["us-gaap"]["Revenues"]["units"]["USD"]
|
|
vals = {(e["end"], e["accn"]): e["val"] for e in out}
|
|
assert vals[("2001-12-31", "bad")] == 2001 * _M
|
|
assert vals[("2011-12-31", "bad")] == 2_500_000
|
|
assert isinstance(vals[("2011-12-31", "bad")], int)
|
|
assert vals[("2001-12-31", "b")] == 2001 * _M
|
|
assert facts["us-gaap"]["Revenues"]["units"]["USD"][0]["val"] == 2001
|
|
|
|
@pytest.mark.parametrize(
|
|
("val", "covers", "expected"),
|
|
[
|
|
(
|
|
351.9,
|
|
(("x", "2019-02-21", 350 * _M), ("y", "2018-05-01", 352 * _M)),
|
|
351_900_000,
|
|
),
|
|
(
|
|
116_238_493,
|
|
(("x", "2019-02-21", 116_238), ("y", "2018-05-01", 116_000)),
|
|
116_238.493,
|
|
),
|
|
(
|
|
351.9,
|
|
(("x", "2019-02-21", 350_000 * _M), ("y", "2018-05-01", 352 * _M)),
|
|
351.9,
|
|
),
|
|
(351.9, (("x", "2019-02-21", 350 * _M),), 351.9),
|
|
],
|
|
)
|
|
def test_share_count_restated_to_cover_page(self, val, covers, expected):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
_SHARES,
|
|
"shares",
|
|
[
|
|
_entry(
|
|
"2017-12-31",
|
|
"2017-01-01",
|
|
filed="2019-02-21",
|
|
val=val,
|
|
accn="x",
|
|
)
|
|
],
|
|
),
|
|
_cover(*covers),
|
|
]
|
|
)
|
|
value = rescaled_facts(facts)["us-gaap"][_SHARES]["units"]["shares"][0]["val"]
|
|
assert value == expected
|
|
assert type(value) is type(expected)
|
|
|
|
def test_disputed_value_takes_concept_scale(self):
|
|
entries = [
|
|
_entry("2020-12-31", "2020-01-01", filed="2021-02-23", val=-26.3, accn="a"),
|
|
_entry(
|
|
"2020-12-31", "2020-01-01", filed="2023-02-24", val=26_300_000, accn="b"
|
|
),
|
|
_entry(
|
|
"2023-12-31", "2023-01-01", filed="2024-02-22", val=71_700_000, accn="c"
|
|
),
|
|
_entry(
|
|
"2020-06-30",
|
|
"2020-04-01",
|
|
form="10-Q",
|
|
filed="2020-08-10",
|
|
val=11_900_000,
|
|
accn="d",
|
|
),
|
|
]
|
|
facts = _facts([("us-gaap", "AssetImpairmentCharges", "USD", entries)])
|
|
out = rescaled_facts(facts)["us-gaap"]["AssetImpairmentCharges"]["units"]["USD"]
|
|
assert [e["val"] for e in out] == [
|
|
-26_300_000,
|
|
26_300_000,
|
|
71_700_000,
|
|
11_900_000,
|
|
]
|
|
|
|
def test_disputed_value_consistent_with_own_filing_kept(self):
|
|
def quarter(year, filed, val, accn):
|
|
return _entry(
|
|
f"{year}-10-31",
|
|
f"{year}-08-01",
|
|
form="10-Q",
|
|
filed=filed,
|
|
val=val,
|
|
accn=accn,
|
|
)
|
|
|
|
entries = [
|
|
quarter(2023, "2023-12-07", 241_490_000, "a"),
|
|
quarter(2022, "2023-12-07", 228_544_000, "a"),
|
|
quarter(2023, "2024-12-05", 241_490, "b"),
|
|
*(quarter(y, f"{y}-12-05", 266_556, f"n{y}") for y in (2024, 2025, 2026)),
|
|
]
|
|
facts = _facts([("us-gaap", "Revenues", "USD", entries)])
|
|
out = rescaled_facts(facts)["us-gaap"]["Revenues"]["units"]["USD"]
|
|
assert [e["val"] for e in out[:3]] == [241_490_000, 228_544_000, 241_490]
|
|
|
|
@pytest.mark.parametrize(
|
|
("entries", "expected"),
|
|
[
|
|
(
|
|
[
|
|
_entry("2020-12-31", "2020-01-01", filed="2021-02-01", val=5),
|
|
_entry("2020-12-31", "2020-01-01", filed="2022-02-01", val=5 * _M),
|
|
_entry("2019-12-31", "2019-01-01", filed="2020-02-01", val=150_000),
|
|
],
|
|
[5, 5 * _M, 150_000],
|
|
),
|
|
(
|
|
[
|
|
_entry("2019-12-31", "2019-10-01", filed="2020-01-31", val=395),
|
|
_entry(
|
|
"2019-12-31", "2019-10-01", filed="2021-02-01", val=395 * _M
|
|
),
|
|
_entry("2018-12-31", "2018-10-01", form="8-K", val=100),
|
|
],
|
|
[395, 395 * _M, 100],
|
|
),
|
|
],
|
|
)
|
|
def test_dispute_without_fitting_scale_unchanged(self, entries, expected):
|
|
facts = _facts([("us-gaap", "Revenues", "USD", entries)])
|
|
out = rescaled_facts(facts)
|
|
assert isinstance(out, ScaledFacts)
|
|
assert out["us-gaap"] is facts["us-gaap"]
|
|
assert [e["val"] for e in out["us-gaap"]["Revenues"]["units"]["USD"]] == (
|
|
expected
|
|
)
|
|
|
|
def test_schema_restates_facts_once(self):
|
|
facts = _facts(
|
|
[("us-gaap", "Revenues", "USD", [_entry("2023-12-31", "2023-01-01")])]
|
|
)
|
|
scaled = StatementSchema._preliminary_facts(facts, False, False)
|
|
assert isinstance(scaled, ScaledFacts)
|
|
assert StatementSchema._preliminary_facts(scaled, False, False) is scaled
|
|
|
|
|
|
class TestComputeRefFilings:
|
|
def test_quarterly_ignores_annual_form_interim_monetary_facts(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-03-31", "2023-01-01", form="10-Q", filed="2023-05-01"
|
|
),
|
|
_entry(
|
|
"2023-03-31", "2023-01-01", form="10-K", filed="2024-02-01"
|
|
),
|
|
],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
ref = compute_ref_filings(facts, [row], "quarterly", "USD")
|
|
assert ref == {"2023-03-31": "2023-05-01"}
|
|
|
|
def test_instant_row_ignores_duration_facts(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Assets",
|
|
"USD",
|
|
[
|
|
_entry("2023-12-31", filed="2024-02-01"),
|
|
_entry("2023-12-31", "2023-01-01", filed="2024-06-01"),
|
|
],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_assets", period_type="instant", xbrl=[("Assets", "us-gaap")])
|
|
ref = compute_ref_filings(facts, [row], "annual", "USD")
|
|
assert ref == {"2023-12-31": "2024-02-01"}
|
|
|
|
def test_prefers_latest_filed_normal_mode(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry("2023-12-31", "2023-01-01", filed="2024-02-01", val=1.0),
|
|
_entry("2023-12-31", "2023-01-01", filed="2024-05-01", val=1.0),
|
|
],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
ref = compute_ref_filings(facts, [row], "annual", "USD")
|
|
assert ref["2023-12-31"] == "2024-05-01" # latest
|
|
|
|
def test_pit_mode_prefers_earliest(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry("2023-12-31", "2023-01-01", filed="2024-02-01", val=1.0),
|
|
_entry("2023-12-31", "2023-01-01", filed="2024-05-01", val=1.0),
|
|
],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
ref = compute_ref_filings(facts, [row], "annual", "USD", pit_mode=True)
|
|
assert ref["2023-12-31"] == "2024-02-01" # earliest
|
|
|
|
def test_stale_filing_gap_excluded(self):
|
|
# filed > 450 days after period end is dropped in non-pit mode.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", filed="2026-01-01", val=1.0)],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert compute_ref_filings(facts, [row], "annual", "USD") == {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# End-to-end on the real BlackRock fixture (annual, quarterly, growth).
|
|
# These exercise the combinatorial extraction/imputation branches that
|
|
# only realistic XBRL data reaches.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def blk_facts():
|
|
with open(_FIXTURE_DIR / "CIK0002012383.json") as f:
|
|
return json.load(f)
|
|
|
|
|
|
class TestBLKEndToEnd:
|
|
def test_annual_pipeline(self, blk_facts):
|
|
res = resolve_company_facts(blk_facts, period="annual")
|
|
assert res.company_type == "diversified"
|
|
assert res.currency == "USD"
|
|
assert len(res.diagnostics) == 0
|
|
# Core identity holds on the latest year.
|
|
ta = next(
|
|
r["value"]
|
|
for r in res.balance_sheet
|
|
if r["tag"] == "total_assets" and r["period_ending"] == "2025-12-31"
|
|
)
|
|
tle = next(
|
|
r["value"]
|
|
for r in res.balance_sheet
|
|
if r["tag"] == "total_liabilities_and_equity"
|
|
and r["period_ending"] == "2025-12-31"
|
|
)
|
|
assert ta == tle
|
|
|
|
def test_quarterly_pipeline_runs_and_aligns(self, blk_facts):
|
|
res = resolve_company_facts(blk_facts, period="quarterly")
|
|
# Quarterly extraction exercises YTD/Q4 derivation at the shared reference vintage.
|
|
assert res.income_statement
|
|
# All three statements share the same set of period-ending dates.
|
|
is_dates = {r["period_ending"] for r in res.income_statement}
|
|
bs_dates = {r["period_ending"] for r in res.balance_sheet}
|
|
cf_dates = {r["period_ending"] for r in res.cash_flow}
|
|
assert is_dates == bs_dates == cf_dates
|
|
|
|
def test_pit_mode_quarterly(self, blk_facts):
|
|
# pit_mode resolves every period at its earliest filing.
|
|
res = resolve_company_facts(blk_facts, period="quarterly", pit_mode=True)
|
|
assert res.income_statement
|
|
|
|
def test_quarterly_skip_imputation_has_no_diagnostics(self, blk_facts):
|
|
results = StatementSchema().extract_all(
|
|
blk_facts, frequency="quarterly", skip_imputation=True
|
|
)
|
|
assert all(not r.diagnostics for r in results.values())
|
|
|
|
def test_both_periods(self, blk_facts):
|
|
res = resolve_company_facts(blk_facts, period="both")
|
|
freqs = {r["frequency"] for r in res.income_statement}
|
|
assert "annual" in freqs and "quarterly" in freqs
|
|
|
|
def test_yoy_growth(self, blk_facts):
|
|
res = resolve_company_facts(blk_facts, period="yoy")
|
|
assert res.income_statement
|
|
assert all(r["unit"] == "percent" for r in res.income_statement)
|
|
|
|
def test_ttm(self, blk_facts):
|
|
res = resolve_company_facts(blk_facts, period="ttm")
|
|
assert all(r["fiscal_period"] == "TTM" for r in res.income_statement)
|
|
|
|
def test_fiscal_year_filter(self, blk_facts):
|
|
res = resolve_company_facts(blk_facts, period="annual", fiscal_years=[2025])
|
|
years = {r["fiscal_year"] for r in res.income_statement}
|
|
assert years == {2025}
|
|
|
|
|
|
class TestSchemaObject:
|
|
"""Direct StatementSchema accessor coverage."""
|
|
|
|
@staticmethod
|
|
@pytest.fixture(scope="class")
|
|
def schema():
|
|
return StatementSchema()
|
|
|
|
def test_version_and_generated(self, schema):
|
|
assert schema.version == "2.0"
|
|
assert schema.generated == "2026-03-17"
|
|
|
|
def test_get_row_and_chain(self, schema):
|
|
row = schema.get_row("total_assets", "balance_sheet", "industrial")
|
|
assert row is not None
|
|
assert row.tag == "total_assets"
|
|
chain = schema.get_tag_chain("total_assets", "balance_sheet", "industrial")
|
|
assert isinstance(chain, tuple) and len(chain) > 0
|
|
|
|
@staticmethod
|
|
def _names(schema, tag, statement, company_type):
|
|
return [
|
|
f"{x['namespace']}:{x['tag']}"
|
|
for x in schema.get_tag_chain(tag, statement, company_type)
|
|
]
|
|
|
|
@pytest.mark.parametrize(
|
|
"company_type", ["industrial", "financial", "diversified", "insurance"]
|
|
)
|
|
def test_chain_scope(self, schema, company_type):
|
|
revenue = self._names(schema, "total_revenue", "income_statement", company_type)
|
|
assert revenue.index("us-gaap:Revenues") < revenue.index(
|
|
"us-gaap:SalesRevenueNet"
|
|
)
|
|
for statement in ("income_statement", "cash_flow"):
|
|
assert "us-gaap:OtherDepreciationAndAmortization" not in self._names(
|
|
schema, "depreciation_and_amortization", statement, company_type
|
|
)
|
|
sbc = self._names(schema, "stock_based_compensation", "cash_flow", company_type)
|
|
assert "us-gaap:StockIssuedDuringPeriodValueShareBasedCompensation" not in sbc
|
|
assert self._names(
|
|
schema, "temporary_equity", "balance_sheet", company_type
|
|
) == [
|
|
"us-gaap:TemporaryEquityCarryingAmountIncludingPortionAttributableToNoncontrollingInterests",
|
|
"us-gaap:TemporaryEquityCarryingAmount",
|
|
]
|
|
rnci = self._names(
|
|
schema, "redeemable_noncontrolling_interest", "balance_sheet", company_type
|
|
)
|
|
assert rnci[-1] == (
|
|
"us-gaap:TemporaryEquityCarryingAmountAttributableToNoncontrollingInterest"
|
|
)
|
|
interest = self._names(
|
|
schema, "total_interest_income", "income_statement", company_type
|
|
)
|
|
assert "us-gaap:InterestIncomeExpenseNet" not in interest
|
|
assert "ifrs-full:InterestRevenueExpense" not in interest
|
|
if company_type == "financial":
|
|
assert not {
|
|
"us-gaap:RevenueFromContractWithCustomerExcludingAssessedTax",
|
|
"us-gaap:RevenueFromContractWithCustomerIncludingAssessedTax",
|
|
"ifrs-full:RevenueFromContractsWithCustomers",
|
|
} & set(revenue)
|
|
if company_type == "insurance":
|
|
investment = self._names(
|
|
schema,
|
|
"investment_securities_interest_income",
|
|
"income_statement",
|
|
company_type,
|
|
)
|
|
assert investment[0] == "us-gaap:NetInvestmentIncome"
|
|
row = schema.get_row(
|
|
"total_interest_income", "income_statement", company_type
|
|
)
|
|
assert row.parent == "investment_securities_interest_income"
|
|
|
|
def test_get_row_missing_returns_none(self, schema):
|
|
assert schema.get_row("does_not_exist", "balance_sheet", "industrial") is None
|
|
assert (
|
|
schema.get_tag_chain("does_not_exist", "balance_sheet", "industrial") == ()
|
|
)
|
|
|
|
def test_get_period_type(self, schema):
|
|
assert (
|
|
schema.get_period_type("total_assets", "balance_sheet", "industrial")
|
|
== "instant"
|
|
)
|
|
assert schema.get_period_type("nope", "balance_sheet", "industrial") is None
|
|
|
|
def test_extract_all_raises_for_annual_only_filer_on_quarterly(self, schema):
|
|
# A 20-F (annual-only) filer with no interim data -> OpenBBError for quarterly.
|
|
from openbb_core.app.model.abstract.error import OpenBBError
|
|
|
|
facts_json = {
|
|
"entityName": "Annual Only Inc.",
|
|
"cik": "0000000001",
|
|
"facts": _facts(
|
|
[
|
|
("ifrs-full", "Assets", "EUR", [_entry("2023-12-31", form="20-F")]),
|
|
(
|
|
"ifrs-full",
|
|
"Revenue",
|
|
"EUR",
|
|
[_entry("2023-12-31", "2023-01-01", form="20-F")],
|
|
),
|
|
]
|
|
),
|
|
}
|
|
with pytest.raises(OpenBBError):
|
|
schema.extract_all(facts_json, frequency="quarterly")
|
|
|
|
def test_merge_facts_combines_units(self, schema):
|
|
a = {
|
|
"facts": _facts(
|
|
[("us-gaap", "Assets", "USD", [_entry("2022-12-31", val=1.0)])]
|
|
)
|
|
}
|
|
b = {
|
|
"facts": _facts(
|
|
[("us-gaap", "Assets", "USD", [_entry("2023-12-31", val=2.0)])]
|
|
)
|
|
}
|
|
merged = schema.merge_facts(a, b)
|
|
assert len(merged["us-gaap"]["Assets"]["units"]["USD"]) == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _detection.py
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetFilingDatesSemiAnnual:
|
|
def test_semi_annual_150_to_200_day_window(self):
|
|
# A 6-K with a ~182-day (H1) duration must register via the
|
|
# 150<=days<=200 quarterly branch (line 127-130).
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenue",
|
|
"USD",
|
|
[_entry("2023-06-30", "2023-01-01", form="6-K")],
|
|
),
|
|
]
|
|
)
|
|
dates = get_filing_dates(facts, "quarterly")
|
|
assert "2023-06-30" in dates
|
|
|
|
def test_preliminary_quarterly_candidate_added(self):
|
|
# 8-K with a ~90 day duration only registers as a preliminary
|
|
# quarterly candidate when include_preliminary=True (line 131-136).
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenue",
|
|
"USD",
|
|
[_entry("2023-09-30", "2023-07-01", form="8-K")],
|
|
),
|
|
]
|
|
)
|
|
with_prelim = get_filing_dates(facts, "quarterly", include_preliminary=True)
|
|
assert "2023-09-30" in with_prelim
|
|
# Without the flag the 8-K quarter is ignored.
|
|
assert "2023-09-30" not in get_filing_dates(facts, "quarterly")
|
|
|
|
|
|
class TestGetFilingDatesCanonicalEdges:
|
|
def test_feb29_anchor_value_error_fallback(self):
|
|
# Dominant month/day is 02-29; for non-leap anchor years datetime()
|
|
# raises ValueError and the min(day, 28) fallback path runs (168-169).
|
|
rev = [
|
|
_entry("2020-02-29", "2019-03-01"),
|
|
_entry("2024-02-29", "2023-03-01"),
|
|
_entry("2016-02-29", "2015-03-01"),
|
|
_entry("2028-02-29", "2027-03-01"),
|
|
]
|
|
assets = [
|
|
_entry("2020-02-29"),
|
|
_entry("2024-02-29"),
|
|
_entry("2016-02-29"),
|
|
_entry("2028-02-29"),
|
|
]
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "Assets", "USD", assets),
|
|
("us-gaap", "Revenues", "USD", rev),
|
|
]
|
|
)
|
|
dates = get_filing_dates(facts, "annual")
|
|
# All four leap-day ends survive the canonical filter.
|
|
assert "2024-02-29" in dates
|
|
assert "2020-02-29" in dates
|
|
|
|
def test_earlier_near_duplicate_dropped_when_later_is_exact(self):
|
|
# Equal support within 10 days: the dominant anchor date is kept.
|
|
rev = [_entry(f"{y}-12-31", f"{y}-01-01") for y in (2021, 2022, 2023, 2024)]
|
|
assets = [_entry(f"{y}-12-31") for y in (2021, 2022, 2023, 2024)]
|
|
# A 52-week fiscal end a few days AFTER a calendar 12-31.
|
|
rev.append(_entry("2025-01-03", "2024-01-05"))
|
|
assets.append(_entry("2025-01-03"))
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "Assets", "USD", assets),
|
|
("us-gaap", "Revenues", "USD", rev),
|
|
]
|
|
)
|
|
dates = get_filing_dates(facts, "annual")
|
|
assert "2024-12-31" in dates
|
|
|
|
def test_later_near_duplicate_dropped_when_earlier_is_exact(self):
|
|
# Equal support within 10 days: the dominant anchor date is kept.
|
|
rev = [_entry(f"{y}-12-31", f"{y}-01-01") for y in (2020, 2021, 2022, 2023)]
|
|
assets = [_entry(f"{y}-12-31") for y in (2020, 2021, 2022, 2023)]
|
|
rev.append(_entry("2024-01-02", "2023-01-04"))
|
|
assets.append(_entry("2024-01-02"))
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "Assets", "USD", assets),
|
|
("us-gaap", "Revenues", "USD", rev),
|
|
]
|
|
)
|
|
dates = get_filing_dates(facts, "annual")
|
|
assert "2023-12-31" in dates
|
|
assert "2024-01-02" not in dates
|
|
|
|
def test_near_duplicate_cluster_keeps_best_supported_date(self):
|
|
# 2023-09-30 carries more annual facts than the 09-24 anchor and 09-25 strays.
|
|
ends = ("2020-09-24", "2021-09-24", "2022-09-24")
|
|
rev = [_entry(e, f"{int(e[:4]) - 1}-09-25") for e in ends]
|
|
ni = [_entry(e, f"{int(e[:4]) - 1}-09-25") for e in ends]
|
|
assets = [_entry(e) for e in ends]
|
|
rev += [
|
|
_entry("2023-09-24", "2022-09-25"),
|
|
_entry("2023-09-25", "2022-09-26"),
|
|
_entry("2023-09-30", "2022-10-01"),
|
|
]
|
|
ni.append(_entry("2023-09-30", "2022-10-01"))
|
|
cogs = [_entry("2023-09-30", "2022-10-01")]
|
|
assets.append(_entry("2023-09-30"))
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "Assets", "USD", assets),
|
|
("us-gaap", "Revenues", "USD", rev),
|
|
("us-gaap", "NetIncomeLoss", "USD", ni),
|
|
("us-gaap", "CostOfRevenue", "USD", cogs),
|
|
]
|
|
)
|
|
dates = get_filing_dates(facts, "annual")
|
|
assert "2023-09-30" in dates
|
|
assert "2023-09-24" not in dates
|
|
assert "2023-09-25" not in dates
|
|
|
|
|
|
class TestGetFiscalMetaEdges:
|
|
def test_annual_entry_without_filed_skipped(self):
|
|
# An annual entry that has no `filed` is skipped (line 266-267);
|
|
# metadata then falls back to int(date[:4]).
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "Assets", "USD", [_entry("2023-12-31", filed="")]),
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", filed="")],
|
|
),
|
|
]
|
|
)
|
|
meta = get_fiscal_meta(facts, "annual", {"2023-12-31"})
|
|
assert meta["2023-12-31"] == {"fiscal_year": 2023, "fiscal_period": "FY"}
|
|
|
|
def test_quarterly_entry_missing_fy_fp_skipped(self):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-09-30", "2023-07-01", form="10-Q", fy=None, fp="")],
|
|
),
|
|
]
|
|
)
|
|
meta = get_fiscal_meta(facts, "quarterly", {"2023-09-30"})
|
|
assert meta["2023-09-30"] == {"fiscal_year": 2023, "fiscal_period": "Q3"}
|
|
|
|
def test_semi_annual_metadata_used_for_quarter(self):
|
|
# A 6-K H1 interim supplies fiscal metadata via the SEMI_ANNUAL
|
|
# branch (279-283) and is read back through best_semi (328-333).
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-06-30",
|
|
"2023-01-01",
|
|
form="6-K",
|
|
fy=2023,
|
|
fp="H1",
|
|
)
|
|
],
|
|
),
|
|
]
|
|
)
|
|
meta = get_fiscal_meta(facts, "quarterly", {"2023-06-30"})
|
|
assert meta["2023-06-30"] == {"fiscal_year": 2023, "fiscal_period": "H1"}
|
|
|
|
def test_preliminary_with_fy_fp_metadata(self):
|
|
# An 8-K carrying explicit fy/fp populates best_preliminary via the
|
|
# first preliminary branch (285-290) and is read back (335-340).
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-09-30",
|
|
"2023-07-01",
|
|
form="8-K",
|
|
fy=2023,
|
|
fp="Q3",
|
|
)
|
|
],
|
|
),
|
|
]
|
|
)
|
|
meta = get_fiscal_meta(facts, "quarterly", {"2023-09-30"})
|
|
assert meta["2023-09-30"] == {"fiscal_year": 2023, "fiscal_period": "Q3"}
|
|
|
|
def test_semi_annual_half_year_classified_as_h1(self):
|
|
# A 6-K half-year (~181-day) interim is classified as H1 from its
|
|
# duration alone; the unreliable fy/fp fields are not required.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-06-30",
|
|
"2023-01-01",
|
|
form="6-K",
|
|
fy=None,
|
|
fp="",
|
|
)
|
|
],
|
|
),
|
|
]
|
|
)
|
|
meta = get_fiscal_meta(facts, "quarterly", {"2023-06-30"})
|
|
assert meta["2023-06-30"] == {"fiscal_year": 2023, "fiscal_period": "H1"}
|
|
|
|
def test_preliminary_without_fy_fp_derives_calendar_quarter(self):
|
|
# An 8-K with no fy/fp derives a calendar quarter from the end
|
|
# month (line 291-298): 2023-09-30 -> Q3.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-09-30",
|
|
"2023-07-01",
|
|
form="8-K",
|
|
fy=None,
|
|
fp="",
|
|
)
|
|
],
|
|
),
|
|
]
|
|
)
|
|
meta = get_fiscal_meta(facts, "quarterly", {"2023-09-30"})
|
|
assert meta["2023-09-30"] == {"fiscal_year": 2023, "fiscal_period": "Q3"}
|
|
|
|
def test_semi_annual_only_annual_date_becomes_h2(self):
|
|
# No 10-Q data, only 6-K interims: an annual date in a quarterly
|
|
# extraction is labelled H2 when non-annual dates do NOT outnumber
|
|
# annual dates (line 312-314).
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Assets",
|
|
"USD",
|
|
[
|
|
_entry("2023-06-30", form="6-K", fy=2023, fp="H1"),
|
|
_entry("2023-12-31", form="10-K", fy=2023, fp="FY"),
|
|
],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-06-30",
|
|
"2023-01-01",
|
|
form="6-K",
|
|
fy=2023,
|
|
fp="H1",
|
|
),
|
|
_entry(
|
|
"2023-12-31",
|
|
"2023-01-01",
|
|
form="10-K",
|
|
fy=2023,
|
|
fp="FY",
|
|
),
|
|
],
|
|
),
|
|
]
|
|
)
|
|
dates = {"2023-06-30", "2023-12-31"}
|
|
meta = get_fiscal_meta(facts, "quarterly", dates)
|
|
# 1 non-annual (H1) is not > 1 annual -> H2.
|
|
assert meta["2023-12-31"]["fiscal_period"] == "H2"
|
|
|
|
def test_quarterly_fiscal_year_from_annual_filing(self):
|
|
# An annual date marked Q4 whose fiscal_year disagrees with the
|
|
# preceding interim's fiscal_year is realigned (line 366-370).
|
|
# Fiscal year ends 2024-01-31; the Q3 interim (2023-10-31) carries
|
|
# fy=2024, the FY end (2024-01-31) is mis-stamped fy=2023.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Assets",
|
|
"USD",
|
|
[
|
|
_entry("2023-10-31", form="10-Q", fy=2024, fp="Q3"),
|
|
_entry("2024-01-31", form="10-K", fy=2023, fp="FY"),
|
|
],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-10-31",
|
|
"2023-08-01",
|
|
form="10-Q",
|
|
fy=2024,
|
|
fp="Q3",
|
|
),
|
|
_entry(
|
|
"2024-01-31",
|
|
"2023-02-01",
|
|
form="10-K",
|
|
fy=2023,
|
|
fp="FY",
|
|
),
|
|
],
|
|
),
|
|
]
|
|
)
|
|
dates = get_filing_dates(facts, "quarterly")
|
|
meta = get_fiscal_meta(facts, "quarterly", dates)
|
|
assert meta["2024-01-31"]["fiscal_year"] == 2023
|
|
assert meta["2024-01-31"]["fiscal_period"] in ("Q4", "H2")
|
|
|
|
|
|
class TestDetectCurrencyTie:
|
|
def test_max_picks_a_currency_on_tie(self):
|
|
# Equal counts of two currencies -> max() still returns one of them
|
|
# (deterministic, exercises the non-empty return at line 398).
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "A", "USD", [_entry("2023-12-31")]),
|
|
("us-gaap", "B", "JPY", [_entry("2023-12-31")]),
|
|
]
|
|
)
|
|
assert detect_reporting_currency(facts) in {"USD", "JPY"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _extraction.py
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetAnnualValuesEdges:
|
|
def test_missing_tag_appends_empty_candidate(self):
|
|
# A tag absent from facts hits the `if not tag_data` continue (72-73);
|
|
# the second (present) tag still resolves.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"RevB",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=200.0)],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("RevA", "us-gaap"), ("RevB", "us-gaap")])
|
|
res = _get_annual_values(facts, row)
|
|
assert res["2023-12-31"][1] == 200.0
|
|
|
|
def test_tag_present_but_no_matching_unit(self):
|
|
# Tag exists but only under a non-monetary unit -> _get_unit_data
|
|
# falls back to first unit; here we force an empty units dict so the
|
|
# `if not unit_data` branch (line 71-73) runs.
|
|
facts = {"us-gaap": {"RevA": {"units": {}}}}
|
|
row = _rd("total_revenue", xbrl=[("RevA", "us-gaap")])
|
|
assert _get_annual_values(facts, row) == {}
|
|
|
|
def test_unparseable_dates_skipped(self):
|
|
# A non-date end/start raises in strptime -> entry skipped (96-97).
|
|
facts = _facts(
|
|
[("us-gaap", "Revenues", "USD", [_entry("bad-date", "worse", val=5.0)])]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert _get_annual_values(facts, row) == {}
|
|
|
|
def test_none_value_skipped(self):
|
|
# An annual-window entry whose val is None is skipped (line 104-105).
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=None)],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert _get_annual_values(facts, row) == {}
|
|
|
|
def test_date_present_only_in_skipped_candidate(self):
|
|
# all_dates collects a date, but every filing for it was filtered so
|
|
# ref_filed stays None -> the `if ref_filed is None: continue` at
|
|
# line 136-137 runs. Achieved by a val=None entry on a lone tag.
|
|
facts = {
|
|
"us-gaap": {
|
|
"Revenues": {
|
|
"units": {
|
|
"USD": [
|
|
{
|
|
"end": "2023-12-31",
|
|
"start": "2023-01-01",
|
|
"val": 10.0,
|
|
"form": "10-K",
|
|
"filed": "2024-02-15",
|
|
}
|
|
]
|
|
}
|
|
}
|
|
}
|
|
}
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
# Sanity: this one resolves.
|
|
assert _get_annual_values(facts, row)["2023-12-31"][1] == 10.0
|
|
|
|
|
|
class TestGetYtd9ValuesEdges:
|
|
def test_missing_tag_continue(self):
|
|
# First tag missing -> `if not tag_data: continue` (168-169); second
|
|
# tag supplies both the annual and the 9-month YTD entries.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"RevB",
|
|
"USD",
|
|
[
|
|
_entry("2023-12-31", "2023-01-01", val=1000.0),
|
|
_entry("2023-09-30", "2023-01-01", val=700.0),
|
|
],
|
|
),
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("RevA", "us-gaap"), ("RevB", "us-gaap")])
|
|
res = _get_ytd9_values(facts, row)
|
|
assert res["2023-12-31"] == 300.0
|
|
|
|
def test_empty_units_continue(self):
|
|
# Tag present but units empty -> `if not unit_data: continue` (173-174).
|
|
facts = {"us-gaap": {"Rev": {"units": {}}}}
|
|
row = _rd("total_revenue", xbrl=[("Rev", "us-gaap")])
|
|
assert _get_ytd9_values(facts, row) == {}
|
|
|
|
def test_zero_length_duration_skipped(self):
|
|
# start == end inside the YTD loop -> skipped (line 180-181); leaves
|
|
# no annual entries so result is empty.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-12-31", val=1000.0)],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert _get_ytd9_values(facts, row) == {}
|
|
|
|
def test_unparseable_and_none_value_skipped(self):
|
|
# Bad date (188-189) and None val (193-194) inside the YTD loop.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry("bad", "worse", val=10.0),
|
|
_entry("2023-12-31", "2023-01-01", val=None),
|
|
],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert _get_ytd9_values(facts, row) == {}
|
|
|
|
def test_no_ytd_match_continue(self):
|
|
# Annual present and a 9-month YTD present but the YTD end does NOT
|
|
# fall strictly inside any (fy_start, fy_end) window, so ytd_end_match
|
|
# stays None -> `if not ytd_end_match: continue` (229-230). The YTD
|
|
# belongs to a different fiscal year than the only annual.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry("2023-12-31", "2023-01-01", val=1000.0),
|
|
# 9-mo YTD ending in a LATER year -> outside 2023 window.
|
|
_entry("2024-09-30", "2024-01-01", val=700.0),
|
|
],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert _get_ytd9_values(facts, row) == {}
|
|
|
|
def test_no_common_filing_uses_latest_each(self):
|
|
# FY and YTD filed on different dates (no common filing) -> the else
|
|
# branch picks latest of each (line 241-243).
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-12-31",
|
|
"2023-01-01",
|
|
filed="2024-02-10",
|
|
val=1000.0,
|
|
),
|
|
_entry(
|
|
"2023-09-30",
|
|
"2023-01-01",
|
|
filed="2023-11-05",
|
|
val=700.0,
|
|
),
|
|
],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
res = _get_ytd9_values(facts, row)
|
|
assert res["2023-12-31"] == 300.0
|
|
|
|
|
|
class TestExtractRowValuesEdges:
|
|
def test_missing_tag_quarterly_appends_empty_ytd_candidate(self):
|
|
# Quarterly duration row with collect_ytd True: a missing first tag
|
|
# appends empty entries to BOTH tag_candidates and ytd_tag_candidates
|
|
# (line 278-284 / 289-294 region).
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"RevB",
|
|
"USD",
|
|
[_entry("2023-03-31", "2023-01-01", form="10-Q", val=100.0)],
|
|
),
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("RevA", "us-gaap"), ("RevB", "us-gaap")])
|
|
vals, _ = extract_row_values(facts, row, "quarterly", "USD")
|
|
assert vals.get("2023-03-31") == 100.0
|
|
|
|
def test_empty_units_quarterly_appends_empty_ytd_candidate(self):
|
|
# Tag present but empty units, quarterly + collect_ytd -> the
|
|
# `if not unit_data` branch appends empty ytd candidate (289-294).
|
|
facts = {
|
|
"us-gaap": {
|
|
"RevA": {"units": {}},
|
|
"RevB": {
|
|
"units": {
|
|
"USD": [
|
|
_entry("2023-03-31", "2023-01-01", form="10-Q", val=100.0)
|
|
]
|
|
}
|
|
},
|
|
}
|
|
}
|
|
row = _rd("total_revenue", xbrl=[("RevA", "us-gaap"), ("RevB", "us-gaap")])
|
|
vals, _ = extract_row_values(facts, row, "quarterly", "USD")
|
|
assert vals.get("2023-03-31") == 100.0
|
|
|
|
def test_entry_without_end_skipped(self):
|
|
# An entry with no `end` is skipped (line 308-309).
|
|
facts = {
|
|
"us-gaap": {
|
|
"Revenues": {
|
|
"units": {
|
|
"USD": [
|
|
{
|
|
"start": "2023-01-01",
|
|
"val": 5.0,
|
|
"form": "10-K",
|
|
"filed": "2024-02-15",
|
|
}
|
|
]
|
|
}
|
|
}
|
|
}
|
|
}
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert extract_row_values(facts, row, "annual", "USD") == ({}, {})
|
|
|
|
def test_unparseable_duration_skipped_annual(self):
|
|
# Bad date in duration branch -> strptime raises -> skipped (323-324).
|
|
facts = _facts(
|
|
[("us-gaap", "Revenues", "USD", [_entry("bad", "worse", val=5.0)])]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert extract_row_values(facts, row, "annual", "USD") == ({}, {})
|
|
|
|
def test_semi_annual_h1_window_quarterly(self):
|
|
# 6-K H1 (~181 days) in quarterly mode registers via the
|
|
# SEMI_ANNUAL_FORMS window branch (line 329-331).
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-06-30", "2023-01-01", form="6-K", val=400.0)],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
vals, _ = extract_row_values(facts, row, "quarterly", "USD")
|
|
assert vals.get("2023-06-30") == 400.0
|
|
|
|
def test_semi_annual_nine_month_extracted(self):
|
|
# A 6-K with a ~273-day (9-month) duration falls in the SEMI_ANNUAL
|
|
# window (240..310) and is extracted as a cumulative interim value.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-09-30", "2023-01-01", form="6-K", val=700.0)],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
vals, _ = extract_row_values(facts, row, "quarterly", "USD")
|
|
assert vals.get("2023-09-30") == 700.0
|
|
|
|
def test_none_value_skipped_after_window(self):
|
|
# A within-window annual entry whose val is None is skipped (348-349).
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=None)],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert extract_row_values(facts, row, "annual", "USD") == ({}, {})
|
|
|
|
def test_quarterly_ytd_derivation_fills_missing_quarter(self):
|
|
# Q1 + Q2 standalone present, Q3 only as a 9-month YTD -> Q3 derived
|
|
# as ytd - prev_cum (line 467-543 YTD reconstruction path), and the
|
|
# ref-resolution `ref is None`/fallback branches (471-493) execute.
|
|
entries = [
|
|
# FY anchor (needed for fy_boundaries).
|
|
_entry("2023-12-31", "2023-01-01", form="10-K", val=1000.0),
|
|
# Standalone quarters from 10-Q vintage.
|
|
_entry("2023-03-31", "2023-01-01", form="10-Q", val=200.0),
|
|
_entry("2023-06-30", "2023-04-01", form="10-Q", val=250.0),
|
|
# 9-month YTD (Jan..Sep) -> 273 days, in the 136..310 YTD band.
|
|
_entry("2023-09-30", "2023-01-01", form="10-Q", val=750.0),
|
|
]
|
|
facts = _facts([("us-gaap", "Revenues", "USD", entries)])
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
vals, srcs = extract_row_values(facts, row, "quarterly", "USD")
|
|
# Q3 derived = 750 (YTD9) - (200 + 250) = 300.
|
|
assert vals.get("2023-09-30") == 300.0
|
|
assert "ytd_derived" in srcs.get("2023-09-30", "")
|
|
|
|
def test_quarterly_first_ytd_past_first_quarter_is_no_quarter(self):
|
|
entries = [
|
|
_entry("2023-12-31", "2023-01-01", form="10-K", val=1000.0),
|
|
_entry("2023-06-30", "2023-01-01", form="10-Q", val=500.0),
|
|
]
|
|
facts = _facts([("us-gaap", "Revenues", "USD", entries)])
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
vals, _ = extract_row_values(facts, row, "quarterly", "USD")
|
|
assert vals == {}
|
|
|
|
@pytest.mark.parametrize(
|
|
("first", "expected_q3"),
|
|
[
|
|
(_entry("2023-06-30", "2023-01-01", form="10-Q", val=500.0), 250.0),
|
|
(_entry("2023-03-31", "2023-01-01", form="10-Q", val=200.0), None),
|
|
(None, None),
|
|
],
|
|
)
|
|
def test_quarterly_ytd_quarter_needs_preceding_quarter(self, first, expected_q3):
|
|
entries = [
|
|
_entry("2023-12-31", "2023-01-01", form="10-K", val=1000.0),
|
|
_entry("2023-09-30", "2023-01-01", form="10-Q", val=750.0),
|
|
*([first] if first else []),
|
|
]
|
|
facts = _facts([("us-gaap", "Revenues", "USD", entries)])
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
vals, srcs = extract_row_values(facts, row, "quarterly", "USD")
|
|
assert "2023-06-30" not in vals
|
|
assert vals.get("2023-09-30") == expected_q3
|
|
assert vals["2023-12-31"] == 250.0
|
|
assert srcs["2023-12-31"] == (
|
|
"Q4: FY[us-gaap:Revenues] \u2212 9M[us-gaap:Revenues]"
|
|
)
|
|
|
|
@pytest.mark.parametrize(("goodwill_fy", "expected"), [(164.0, 0.0), (None, None)])
|
|
def test_q4_from_nine_months_on_one_concept(self, goodwill_fy, expected):
|
|
goodwill = [
|
|
_entry("2025-09-30", "2025-01-01", form="10-Q", val=164.0),
|
|
_entry("2025-09-30", "2025-07-01", form="10-Q", val=164.0),
|
|
]
|
|
if goodwill_fy is not None:
|
|
goodwill.append(_entry("2025-12-31", "2025-01-01", val=goodwill_fy))
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"AssetImpairmentCharges",
|
|
"USD",
|
|
[_entry("2025-12-31", "2025-01-01", val=26.0)],
|
|
),
|
|
("us-gaap", "GoodwillImpairmentLoss", "USD", goodwill),
|
|
]
|
|
)
|
|
row = _rd(
|
|
"impairment_expense",
|
|
xbrl=[
|
|
("AssetImpairmentCharges", "us-gaap"),
|
|
("GoodwillImpairmentLoss", "us-gaap"),
|
|
],
|
|
)
|
|
vals, srcs = extract_row_values(facts, row, "quarterly", "USD")
|
|
assert vals.get("2025-12-31") == expected
|
|
if expected is not None:
|
|
assert srcs["2025-12-31"] == (
|
|
"Q4: FY[us-gaap:GoodwillImpairmentLoss] \u2212 "
|
|
"9M[us-gaap:GoodwillImpairmentLoss]"
|
|
)
|
|
|
|
def test_quarterly_ytd_with_shifted_start_uses_prior_quarters(self):
|
|
entries = [
|
|
_entry("2023-12-31", "2023-01-01", form="10-K", val=1220.0),
|
|
_entry("2023-03-31", "2023-01-01", form="10-Q", val=305.0),
|
|
_entry("2023-06-30", "2023-01-03", form="10-Q", val=610.0),
|
|
_entry("2023-09-30", "2023-01-01", form="10-Q", val=915.0),
|
|
]
|
|
facts = _facts([("us-gaap", "Revenues", "USD", entries)])
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
vals, srcs = extract_row_values(facts, row, "quarterly", "USD")
|
|
assert vals["2023-06-30"] == 305.0
|
|
assert vals["2023-09-30"] == 305.0
|
|
assert vals["2023-12-31"] == 305.0
|
|
assert srcs["2023-09-30"] == "ytd_derived(us-gaap:Revenues)"
|
|
|
|
def test_quarterly_ytd_with_ref_filed_map_fallback(self):
|
|
# Same YTD reconstruction but driving the ref_filed_map branch where
|
|
# the requested ref is absent for the YTD date -> the
|
|
# `before/min(ytd_fdata)` fallback (line 492-494) runs.
|
|
entries = [
|
|
_entry(
|
|
"2023-12-31", "2023-01-01", form="10-K", filed="2024-02-15", val=1000.0
|
|
),
|
|
_entry(
|
|
"2023-03-31", "2023-01-01", form="10-Q", filed="2023-05-01", val=200.0
|
|
),
|
|
_entry(
|
|
"2023-06-30", "2023-04-01", form="10-Q", filed="2023-08-01", val=250.0
|
|
),
|
|
_entry(
|
|
"2023-09-30", "2023-01-01", form="10-Q", filed="2023-11-01", val=750.0
|
|
),
|
|
]
|
|
facts = _facts([("us-gaap", "Revenues", "USD", entries)])
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
# ref_filed_map points the YTD date at a filing that does not exist.
|
|
ref_map = {
|
|
"2023-03-31": "2023-05-01",
|
|
"2023-06-30": "2023-08-01",
|
|
"2023-09-30": "2099-01-01",
|
|
"2023-12-31": "2024-02-15",
|
|
}
|
|
vals, srcs = extract_row_values(
|
|
facts, row, "quarterly", "USD", ref_filed_map=ref_map
|
|
)
|
|
assert vals.get("2023-09-30") == 300.0
|
|
|
|
def test_quarterly_ytd_ref_map_missing_date_skips_resolution(self):
|
|
# ref_filed_map omits the YTD date entirely -> `ref = map.get(date)`
|
|
# is None and the YTD candidate's `if ref is None: continue` (480-481)
|
|
# runs, so no ytd_derived value is produced for that date.
|
|
entries = [
|
|
_entry(
|
|
"2023-12-31", "2023-01-01", form="10-K", filed="2024-02-15", val=1000.0
|
|
),
|
|
_entry(
|
|
"2023-03-31", "2023-01-01", form="10-Q", filed="2023-05-01", val=200.0
|
|
),
|
|
_entry(
|
|
"2023-06-30", "2023-04-01", form="10-Q", filed="2023-08-01", val=250.0
|
|
),
|
|
_entry(
|
|
"2023-09-30", "2023-01-01", form="10-Q", filed="2023-11-01", val=750.0
|
|
),
|
|
]
|
|
facts = _facts([("us-gaap", "Revenues", "USD", entries)])
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
# Deliberately leave 2023-09-30 (the YTD date) out of the map.
|
|
ref_map = {
|
|
"2023-03-31": "2023-05-01",
|
|
"2023-06-30": "2023-08-01",
|
|
"2023-12-31": "2024-02-15",
|
|
}
|
|
vals, srcs = extract_row_values(
|
|
facts, row, "quarterly", "USD", ref_filed_map=ref_map
|
|
)
|
|
# Q3 was not resolved via the YTD path (its ref was None).
|
|
assert "ytd_derived" not in srcs.get("2023-09-30", "")
|
|
|
|
|
|
class TestComputeRefFilingsEdges:
|
|
def test_missing_tag_continue(self):
|
|
# A row tag absent from facts -> `if not tag_data: continue` (614).
|
|
# Provided indirectly: row chains two tags, first missing.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"RevB",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", filed="2024-02-01")],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("RevA", "us-gaap"), ("RevB", "us-gaap")])
|
|
ref = compute_ref_filings(facts, [row], "annual", "USD")
|
|
assert ref["2023-12-31"] == "2024-02-01"
|
|
|
|
def test_empty_units_continue(self):
|
|
# Tag present but empty units -> `if not unit_data: continue` (619-620).
|
|
facts = {
|
|
"us-gaap": {
|
|
"RevA": {"units": {}},
|
|
"RevB": {
|
|
"units": {
|
|
"USD": [_entry("2023-12-31", "2023-01-01", filed="2024-02-01")]
|
|
}
|
|
},
|
|
}
|
|
}
|
|
row = _rd("total_revenue", xbrl=[("RevA", "us-gaap"), ("RevB", "us-gaap")])
|
|
ref = compute_ref_filings(facts, [row], "annual", "USD")
|
|
assert ref["2023-12-31"] == "2024-02-01"
|
|
|
|
def test_entry_without_end_skipped(self):
|
|
# Entry lacking `end` -> skipped at line 630-631.
|
|
facts = {
|
|
"us-gaap": {
|
|
"Revenues": {
|
|
"units": {
|
|
"USD": [
|
|
{
|
|
"start": "2023-01-01",
|
|
"val": 1.0,
|
|
"form": "10-K",
|
|
"filed": "2024-02-01",
|
|
}
|
|
]
|
|
}
|
|
}
|
|
}
|
|
}
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert compute_ref_filings(facts, [row], "annual", "USD") == {}
|
|
|
|
def test_unparseable_duration_skipped(self):
|
|
# Bad date in duration branch -> strptime raises -> skipped (644-645).
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("bad", "worse", filed="2024-02-01")],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert compute_ref_filings(facts, [row], "annual", "USD") == {}
|
|
|
|
def test_semi_annual_h1_window_quarterly(self):
|
|
# 6-K H1 (~181 days) accepted via SEMI_ANNUAL window (line 650-652).
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-06-30", "2023-01-01", form="6-K", filed="2023-08-01"
|
|
)
|
|
],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
ref = compute_ref_filings(facts, [row], "quarterly", "USD")
|
|
assert ref["2023-06-30"] == "2023-08-01"
|
|
|
|
def test_semi_annual_nine_month_produces_ref(self):
|
|
# A 6-K with a ~273-day (9-month) duration falls in the SEMI_ANNUAL
|
|
# window (240..310) and contributes a reference filing.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-09-30", "2023-01-01", form="6-K", filed="2023-11-01"
|
|
)
|
|
],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert compute_ref_filings(facts, [row], "quarterly", "USD") == {
|
|
"2023-09-30": "2023-11-01"
|
|
}
|
|
|
|
def test_out_of_window_quarterly_duration_skipped(self):
|
|
# A 10-Q-vintage duration of ~273 days is neither 60..135 (quarter)
|
|
# nor a semi-annual form -> rejected by `elif not 60<=days<=135`
|
|
# (line 653-654) leaving no ref.
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-09-30", "2023-01-01", form="10-Q", filed="2023-11-01"
|
|
)
|
|
],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert compute_ref_filings(facts, [row], "quarterly", "USD") == {}
|
|
|
|
def test_entry_without_filed_skipped(self):
|
|
# An entry with no `filed` -> skipped (line 658-659).
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", filed="")],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert compute_ref_filings(facts, [row], "annual", "USD") == {}
|
|
|
|
def test_unparseable_filed_gap_skipped(self):
|
|
# A non-pit-mode entry whose `filed` is unparseable raises in the gap
|
|
# computation -> skipped (line 667-668).
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", filed="not-a-date")],
|
|
)
|
|
]
|
|
)
|
|
row = _rd("total_revenue", xbrl=[("Revenues", "us-gaap")])
|
|
assert compute_ref_filings(facts, [row], "annual", "USD") == {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _schema.py -- drive StatementSchema.extract() directly.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def schema():
|
|
return StatementSchema()
|
|
|
|
|
|
def test_combine_unfunded_provision_skips_none_add(schema):
|
|
row = RowResult(
|
|
tag="provision_for_credit_losses",
|
|
label="Provision for credit losses",
|
|
description="",
|
|
parent=None,
|
|
sequence=1,
|
|
factor="+",
|
|
balance="debit",
|
|
unit="monetary",
|
|
period_type="duration",
|
|
values={"2024-12-31": 100.0},
|
|
sources={
|
|
"2024-12-31": "us-gaap:FinancingReceivableExcludingAccruedInterestCreditLossExpenseReversal"
|
|
},
|
|
)
|
|
with patch(
|
|
"openbb_sec.utils.statement_schema._schema.extract_row_values",
|
|
return_value=({}, {}),
|
|
):
|
|
schema._combine_unfunded_provision(
|
|
[row],
|
|
"income_statement",
|
|
"financial",
|
|
{"us-gaap": {}},
|
|
"annual",
|
|
"USD",
|
|
{},
|
|
)
|
|
assert row.values["2024-12-31"] == 100.0
|
|
|
|
|
|
class TestSchemaExtractDefaults:
|
|
def test_extract_infers_type_dates_and_ref_map(self, schema):
|
|
# Calling extract() with company_type / filing_dates / ref_filed_map
|
|
# all None drives the inference branches (line 229-235, 246-254).
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", form="10-K", val=500.0 * _M)],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"CostsAndExpenses",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", form="10-K", val=300.0 * _M)],
|
|
),
|
|
]
|
|
)
|
|
res = schema.extract({"facts": facts}, "income_statement", frequency="annual")
|
|
assert res.statement == "income_statement"
|
|
assert res.currency == "USD"
|
|
assert "2023-12-31" in res.dates
|
|
|
|
def test_extract_quarterly_computes_annual_ref_map(self, schema):
|
|
# Quarterly extract() with no maps computes the annual reference map for Q4.
|
|
entries = [
|
|
_entry("2023-03-31", "2023-01-01", form="10-Q", val=100.0 * _M),
|
|
_entry("2023-06-30", "2023-04-01", form="10-Q", val=150.0 * _M),
|
|
_entry("2023-09-30", "2023-07-01", form="10-Q", val=200.0 * _M),
|
|
_entry("2023-12-31", "2023-01-01", form="10-K", val=600.0 * _M),
|
|
]
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "Revenues", "USD", entries),
|
|
(
|
|
"us-gaap",
|
|
"Assets",
|
|
"USD",
|
|
[
|
|
_entry("2023-03-31", form="10-Q"),
|
|
_entry("2023-06-30", form="10-Q"),
|
|
_entry("2023-09-30", form="10-Q"),
|
|
_entry("2023-12-31", form="10-K"),
|
|
],
|
|
),
|
|
]
|
|
)
|
|
res = schema.extract(
|
|
{"facts": facts}, "income_statement", frequency="quarterly"
|
|
)
|
|
rev = next(r for r in res.rows if r.tag == "total_revenue")
|
|
assert rev.values["2023-12-31"] == 150.0 * _M
|
|
assert rev.sources["2023-12-31"].startswith("Q4: FY[us-gaap:Revenues]")
|
|
|
|
def test_extract_include_preliminary_marks_dates(self, schema):
|
|
# include_preliminary populates _preliminary_dates and tags sources
|
|
# with the "preliminary:" prefix (line 237-240, 431-438).
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2024-12-31",
|
|
"2024-01-01",
|
|
form="10-K",
|
|
filed="2025-02-15",
|
|
val=500.0 * _M,
|
|
),
|
|
_entry(
|
|
"2025-12-31",
|
|
"2025-01-01",
|
|
form="8-K",
|
|
filed="2026-01-30",
|
|
val=600.0 * _M,
|
|
),
|
|
],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"CostsAndExpenses",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2024-12-31",
|
|
"2024-01-01",
|
|
form="10-K",
|
|
filed="2025-02-15",
|
|
val=300.0 * _M,
|
|
),
|
|
_entry(
|
|
"2025-12-31",
|
|
"2025-01-01",
|
|
form="8-K",
|
|
filed="2026-01-30",
|
|
val=350.0 * _M,
|
|
),
|
|
],
|
|
),
|
|
]
|
|
)
|
|
res = schema.extract(
|
|
{"facts": facts},
|
|
"income_statement",
|
|
frequency="annual",
|
|
include_preliminary=True,
|
|
)
|
|
# The 8-K-only 2025 date is preliminary.
|
|
assert "2025-12-31" in res.preliminary_dates
|
|
rev = next(r for r in res.rows if r.tag == "total_revenue")
|
|
assert rev.sources["2025-12-31"].startswith("preliminary:")
|
|
|
|
|
|
class TestSchemaExtractCashFlow:
|
|
def _cf_facts(self, *, nc_val, eop_vals):
|
|
"""Build cash-flow facts: 3 activities + fx + net_change + EOP cash.
|
|
|
|
``eop_vals`` maps end-date -> EOP cash instant value.
|
|
"""
|
|
op = "NetCashProvidedByUsedInOperatingActivities"
|
|
inv = "NetCashProvidedByUsedInInvestingActivities"
|
|
fin = "NetCashProvidedByUsedInFinancingActivities"
|
|
fx = "EffectOfExchangeRateOnCashAndCashEquivalents"
|
|
ncd = "CashAndCashEquivalentsPeriodIncreaseDecrease"
|
|
eop = "CashCashEquivalentsRestrictedCashAndRestrictedCashEquivalents"
|
|
tags = []
|
|
for d in eop_vals:
|
|
y = d[:4]
|
|
start = f"{y}-01-01"
|
|
tags.extend(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
op,
|
|
"USD",
|
|
[_entry(d, start, form="10-K", val=500.0 * _M)],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
inv,
|
|
"USD",
|
|
[_entry(d, start, form="10-K", val=-200.0 * _M)],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
fin,
|
|
"USD",
|
|
[_entry(d, start, form="10-K", val=-150.0 * _M)],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
fx,
|
|
"USD",
|
|
[_entry(d, start, form="10-K", val=50.0 * _M)],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
ncd,
|
|
"USD",
|
|
[_entry(d, start, form="10-K", val=nc_val)],
|
|
),
|
|
("us-gaap", eop, "USD", [_entry(d, form="10-K", val=eop_vals[d])]),
|
|
# Assets instant anchors the annual date resolution.
|
|
("us-gaap", "Assets", "USD", [_entry(d, form="10-K")]),
|
|
]
|
|
)
|
|
return _facts(tags)
|
|
|
|
def test_eop_standalone_backfill_and_bop_derivation(self, schema):
|
|
# Two consecutive years with EOP cash instants: the second year's BOP
|
|
# is derived from the prior year's EOP (line 347-358) and the EOP
|
|
# standalone backfill (line 336-345) populates eop from facts.
|
|
facts = self._cf_facts(
|
|
nc_val=200.0 * _M,
|
|
eop_vals={"2022-12-31": 1000.0 * _M, "2023-12-31": 1200.0 * _M},
|
|
)
|
|
res = schema.extract({"facts": facts}, "cash_flow", frequency="annual")
|
|
bop = next(
|
|
(r for r in res.rows if r.tag == "cash_at_beginning_of_period"), None
|
|
)
|
|
assert bop is not None
|
|
# BOP(2023) derived from EOP(2022) = 1000M.
|
|
assert bop.values.get("2023-12-31") == 1000.0 * _M
|
|
assert "derived: cash_at_end_of_period" in bop.sources["2023-12-31"]
|
|
|
|
def test_identity_enforced_on_bop_when_derived(self, schema):
|
|
# EOP - net_change != BOP(derived): because BOP source contains
|
|
# "derived:", BOP is rewritten to EOP - net_change (line 382-387).
|
|
# Year 1 EOP=1000, Year 2 EOP=1500, net_change=200 -> derived BOP
|
|
# (1000) violates 1000 + 200 != 1500 -> enforced to 1500-200=1300.
|
|
facts = self._cf_facts(
|
|
nc_val=200.0 * _M,
|
|
eop_vals={"2022-12-31": 1000.0 * _M, "2023-12-31": 1500.0 * _M},
|
|
)
|
|
res = schema.extract({"facts": facts}, "cash_flow", frequency="annual")
|
|
bop = next(r for r in res.rows if r.tag == "cash_at_beginning_of_period")
|
|
assert bop.values["2023-12-31"] == 1300.0 * _M
|
|
assert "identity-enforced" in bop.sources["2023-12-31"]
|
|
|
|
def test_net_change_corrected_when_activities_explain_cash_change(self, schema):
|
|
facts = self._cf_facts(
|
|
nc_val=999.0 * _M,
|
|
eop_vals={"2022-12-31": 1000.0 * _M, "2023-12-31": 1200.0 * _M},
|
|
)
|
|
res = schema.extract({"facts": facts}, "cash_flow", frequency="annual")
|
|
nc = next(r for r in res.rows if r.tag == "net_change_in_cash")
|
|
assert nc.values["2023-12-31"] == 200.0 * _M
|
|
assert nc.sources["2023-12-31"] == (
|
|
"corrected: cash_at_end_of_period - cash_at_beginning_of_period"
|
|
)
|
|
other = next(
|
|
(r for r in res.rows if r.tag == "other_net_changes_in_cash"), None
|
|
)
|
|
assert other is None or "2023-12-31" not in other.values
|
|
bop = next(r for r in res.rows if r.tag == "cash_at_beginning_of_period")
|
|
assert bop.values["2023-12-31"] == 1000.0 * _M
|
|
|
|
|
|
class TestSchemaExtractBalanceSheetAnchor:
|
|
def _facts(self, *, with_le):
|
|
tags = [
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2022-12-31", "2022-01-01", filed="2023-02-15", val=900.0 * _M
|
|
)
|
|
],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"Assets",
|
|
"USD",
|
|
[
|
|
_entry("2022-12-31", filed="2023-02-15", val=300.0 * _M),
|
|
_entry("2022-12-31", filed="2024-02-15", val=300.0 * _M),
|
|
],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"Liabilities",
|
|
"USD",
|
|
[_entry("2022-12-31", filed="2023-02-15", val=200.0 * _M)],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"StockholdersEquity",
|
|
"USD",
|
|
[
|
|
_entry("2022-12-31", filed="2023-02-15", val=100.0 * _M),
|
|
_entry("2022-12-31", filed="2024-02-15", val=90.0 * _M),
|
|
],
|
|
),
|
|
]
|
|
if with_le:
|
|
tags.append(
|
|
(
|
|
"us-gaap",
|
|
"LiabilitiesAndStockholdersEquity",
|
|
"USD",
|
|
[_entry("2022-12-31", filed="2023-02-15", val=300.0 * _M)],
|
|
)
|
|
)
|
|
return _facts(tags)
|
|
|
|
def _row(self, res, tag):
|
|
return next(r for r in res.rows if r.tag == tag)
|
|
|
|
def test_liabilities_and_equity_filing_is_reference(self, schema):
|
|
res = schema.extract(
|
|
{"facts": self._facts(with_le=True)},
|
|
"balance_sheet",
|
|
frequency="annual",
|
|
company_type="industrial",
|
|
)
|
|
te = self._row(res, "total_equity")
|
|
assert te.values["2022-12-31"] == 100.0 * _M
|
|
assert te.sources["2022-12-31"] == "us-gaap:StockholdersEquity"
|
|
assert self._row(res, "total_liabilities").sources["2022-12-31"] == (
|
|
"us-gaap:Liabilities"
|
|
)
|
|
|
|
def test_anchor_overrides_supplied_reference_map(self, schema):
|
|
res = schema.extract(
|
|
{"facts": self._facts(with_le=True)},
|
|
"balance_sheet",
|
|
frequency="annual",
|
|
company_type="industrial",
|
|
ref_filed_map={"2022-12-31": "2024-02-15"},
|
|
)
|
|
assert self._row(res, "total_equity").values["2022-12-31"] == 100.0 * _M
|
|
|
|
def test_total_assets_filing_is_reference_without_liabilities_and_equity(
|
|
self, schema
|
|
):
|
|
res = schema.extract(
|
|
{"facts": self._facts(with_le=False)},
|
|
"balance_sheet",
|
|
frequency="annual",
|
|
company_type="industrial",
|
|
)
|
|
te = self._row(res, "total_equity")
|
|
assert te.values["2022-12-31"] == 90.0 * _M
|
|
assert te.sources["2022-12-31"] == "us-gaap:StockholdersEquity"
|
|
assert self._row(res, "total_liabilities").sources["2022-12-31"] == (
|
|
"us-gaap:Liabilities(fallback)"
|
|
)
|
|
|
|
|
|
class TestSchemaExtractChainSubsets:
|
|
def _facts(self, broad, narrow):
|
|
return _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=900.0 * _M)],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"ResearchAndDevelopmentExpense",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=broad)],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"ResearchAndDevelopmentExpenseExcludingAcquiredInProcessCost",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=narrow)],
|
|
),
|
|
]
|
|
)
|
|
|
|
def _rd_row(self, schema, facts):
|
|
res = schema.extract(
|
|
{"facts": facts},
|
|
"income_statement",
|
|
frequency="annual",
|
|
company_type="industrial",
|
|
)
|
|
return next(r for r in res.rows if r.tag == "rd_expense")
|
|
|
|
def test_larger_narrower_concept_replaces_partial_broader_value(self, schema):
|
|
rd = self._rd_row(schema, self._facts(100.0 * _M, 150.0 * _M))
|
|
assert rd.values["2023-12-31"] == 150.0 * _M
|
|
assert rd.sources["2023-12-31"] == (
|
|
"us-gaap:ResearchAndDevelopmentExpenseExcludingAcquiredInProcessCost"
|
|
)
|
|
|
|
def test_smaller_narrower_concept_keeps_broader_value(self, schema):
|
|
rd = self._rd_row(schema, self._facts(120.0 * _M, 110.0 * _M))
|
|
assert rd.values["2023-12-31"] == 120.0 * _M
|
|
assert rd.sources["2023-12-31"] == "us-gaap:ResearchAndDevelopmentExpense"
|
|
|
|
@pytest.mark.parametrize(
|
|
("revenues", "sales", "expected", "source"),
|
|
[
|
|
(0.0, 950.0 * _M, 950.0 * _M, "us-gaap:SalesRevenueNet"),
|
|
(1000.0 * _M, 950.0 * _M, 1000.0 * _M, "us-gaap:Revenues"),
|
|
],
|
|
)
|
|
def test_revenue_subset(self, schema, revenues, sales, expected, source):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=revenues)],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"SalesRevenueNet",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=sales)],
|
|
),
|
|
]
|
|
)
|
|
res = schema.extract(
|
|
{"facts": facts},
|
|
"income_statement",
|
|
frequency="annual",
|
|
company_type="industrial",
|
|
)
|
|
row = next(r for r in res.rows if r.tag == "total_revenue")
|
|
assert row.values["2023-12-31"] == expected
|
|
assert row.sources["2023-12-31"] == source
|
|
|
|
@pytest.mark.parametrize(
|
|
("goods", "cost", "expected", "source"),
|
|
[
|
|
(33 * _M, 40199 * _M, 40199 * _M, "us-gaap:CostOfRevenue"),
|
|
(900 * _M, 800 * _M, 900 * _M, "us-gaap:CostOfGoodsAndServicesSold"),
|
|
],
|
|
)
|
|
def test_cost_of_revenue_subset(self, schema, goods, cost, expected, source):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"CostOfGoodsAndServicesSold",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=goods)],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"CostOfRevenue",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", val=cost)],
|
|
),
|
|
]
|
|
)
|
|
res = schema.extract(
|
|
{"facts": facts},
|
|
"income_statement",
|
|
frequency="annual",
|
|
company_type="industrial",
|
|
)
|
|
row = next(r for r in res.rows if r.tag == "total_cost_of_revenue")
|
|
assert row.values["2023-12-31"] == expected
|
|
assert row.sources["2023-12-31"] == source
|
|
|
|
def test_nonnegative_concept_reads_as_magnitude(self, schema):
|
|
entries = [
|
|
_entry("2023-03-31", "2023-01-01", form="10-Q", val=-100.0 * _M),
|
|
_entry("2023-06-30", "2023-04-01", form="10-Q", val=-110.0 * _M),
|
|
_entry("2023-06-30", "2023-01-01", form="10-Q", val=-210.0 * _M),
|
|
_entry("2023-09-30", "2023-07-01", form="10-Q", val=-120.0 * _M),
|
|
_entry("2023-12-31", "2023-01-01", form="10-K", val=-450.0 * _M),
|
|
]
|
|
facts = _facts(
|
|
[
|
|
("us-gaap", "ResearchAndDevelopmentExpense", "USD", entries),
|
|
(
|
|
"us-gaap",
|
|
"Assets",
|
|
"USD",
|
|
[
|
|
_entry("2023-03-31", form="10-Q"),
|
|
_entry("2023-06-30", form="10-Q"),
|
|
_entry("2023-09-30", form="10-Q"),
|
|
_entry("2023-12-31", form="10-K"),
|
|
],
|
|
),
|
|
]
|
|
)
|
|
quarterly = schema.extract(
|
|
{"facts": facts}, "income_statement", frequency="quarterly"
|
|
)
|
|
rd = next(r for r in quarterly.rows if r.tag == "rd_expense")
|
|
assert [rd.values[d] for d in sorted(rd.values)] == [
|
|
100.0 * _M,
|
|
110.0 * _M,
|
|
120.0 * _M,
|
|
120.0 * _M,
|
|
]
|
|
annual = schema.extract(
|
|
{"facts": facts}, "income_statement", frequency="annual"
|
|
)
|
|
rd = next(r for r in annual.rows if r.tag == "rd_expense")
|
|
assert rd.values["2023-12-31"] == 450.0 * _M
|
|
|
|
def test_net_position_depreciation_negative_not_taken(self, schema):
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"DepreciationDepletionAndAmortization",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2022-12-31",
|
|
"2022-01-01",
|
|
filed="2023-02-15",
|
|
val=-500.0 * _M,
|
|
),
|
|
_entry(
|
|
"2022-12-31",
|
|
"2022-01-01",
|
|
filed="2024-06-15",
|
|
val=500.0 * _M,
|
|
),
|
|
_entry("2023-03-31", "2023-01-01", form="10-Q", val=100.0 * _M),
|
|
_entry(
|
|
"2023-06-30", "2023-01-01", form="10-Q", val=-210.0 * _M
|
|
),
|
|
],
|
|
),
|
|
]
|
|
)
|
|
annual = schema.extract(
|
|
{"facts": facts}, "income_statement", frequency="annual"
|
|
)
|
|
da = next(r for r in annual.rows if r.tag == "depreciation_and_amortization")
|
|
assert da.values["2022-12-31"] == 500.0 * _M
|
|
quarterly = schema.extract(
|
|
{"facts": facts}, "income_statement", frequency="quarterly"
|
|
)
|
|
da = next(r for r in quarterly.rows if r.tag == "depreciation_and_amortization")
|
|
assert da.values["2023-03-31"] == 100.0 * _M
|
|
assert "2023-06-30" not in da.values
|
|
|
|
def test_eop_stale_filing_recovered_via_instant_fallback(self, schema):
|
|
# The EOP tag's only filing is stale (filed > 450 days after the
|
|
# period end) so compute_ref_filings excludes it from the shared ref
|
|
# map; the instant fallback in extract_row_values still recovers the
|
|
# value via min(filings). Exercises the cash_flow EOP/BOP wiring with
|
|
# a ref map driven by a separate (non-stale) operating-activity tag.
|
|
op = "NetCashProvidedByUsedInOperatingActivities"
|
|
eop = "CashCashEquivalentsRestrictedCashAndRestrictedCashEquivalents"
|
|
facts = _facts(
|
|
[
|
|
# Operating activity (non-stale) anchors both years' ref map.
|
|
(
|
|
"us-gaap",
|
|
op,
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2022-12-31",
|
|
"2022-01-01",
|
|
form="10-K",
|
|
filed="2023-02-15",
|
|
val=400.0 * _M,
|
|
),
|
|
_entry(
|
|
"2023-12-31",
|
|
"2023-01-01",
|
|
form="10-K",
|
|
filed="2024-02-15",
|
|
val=500.0 * _M,
|
|
),
|
|
],
|
|
),
|
|
# EOP cash for 2023 filed > 450 days late -> dropped from the
|
|
# shared ref map but recoverable standalone.
|
|
(
|
|
"us-gaap",
|
|
eop,
|
|
"USD",
|
|
[
|
|
_entry(
|
|
"2023-12-31",
|
|
form="10-K",
|
|
filed="2025-06-01",
|
|
val=1200.0 * _M,
|
|
)
|
|
],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"Assets",
|
|
"USD",
|
|
[
|
|
_entry("2022-12-31", form="10-K", filed="2023-02-15"),
|
|
_entry("2023-12-31", form="10-K", filed="2024-02-15"),
|
|
],
|
|
),
|
|
]
|
|
)
|
|
res = schema.extract({"facts": facts}, "cash_flow", frequency="annual")
|
|
eop_row = next((r for r in res.rows if r.tag == "cash_at_end_of_period"), None)
|
|
# The stale 2023 EOP was backfilled via the standalone extraction.
|
|
assert eop_row is not None
|
|
assert eop_row.values.get("2023-12-31") == 1200.0 * _M
|
|
|
|
def test_eop_standalone_backfill_when_date_absent_from_ref_map(self, schema):
|
|
# A date present in ``filing_dates`` but absent from the supplied
|
|
# ``ref_filed_map`` is skipped by the main per-row extraction (ref_filed
|
|
# resolves to None), so the standalone EOP backfill (lines 342-345)
|
|
# recovers the value straight from the instant fact.
|
|
eop = "CashCashEquivalentsRestrictedCashAndRestrictedCashEquivalents"
|
|
facts = _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
eop,
|
|
"USD",
|
|
[_entry("2023-12-31", form="10-K", val=1200.0 * _M)],
|
|
)
|
|
]
|
|
)
|
|
res = schema.extract(
|
|
{"facts": facts},
|
|
"cash_flow",
|
|
frequency="annual",
|
|
filing_dates={"2023-12-31"},
|
|
ref_filed_map={},
|
|
)
|
|
eop_row = next(r for r in res.rows if r.tag == "cash_at_end_of_period")
|
|
assert eop_row.values["2023-12-31"] == 1200.0 * _M
|
|
assert "CashCashEquivalents" in eop_row.sources["2023-12-31"]
|
|
|
|
|
|
class TestSchemaQuarterlyFiscalYearEnd:
|
|
def test_fiscal_year_end_kept_when_income_statement_lacks_it(self, schema):
|
|
m = 1_000_000
|
|
quarters = ("2023-03-31", "2023-06-30", "2023-09-30")
|
|
instants = ("2022-12-31", *quarters, "2023-12-31")
|
|
facts_json = {
|
|
"entityName": "Shifted Quarter Inc.",
|
|
"facts": _facts(
|
|
[
|
|
(
|
|
"us-gaap",
|
|
"Assets",
|
|
"USD",
|
|
[
|
|
_entry(d, form="10-K" if d.endswith("12-31") else "10-Q")
|
|
for d in instants
|
|
],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"CashAndCashEquivalentsAtCarryingValue",
|
|
"USD",
|
|
[
|
|
_entry(
|
|
d,
|
|
form="10-K" if d.endswith("12-31") else "10-Q",
|
|
val=10.0 * m,
|
|
)
|
|
for d in instants
|
|
],
|
|
),
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[
|
|
_entry("2022-12-31", "2022-01-01", val=800.0 * m),
|
|
_entry("2023-12-31", "2023-01-01", val=1000.0 * m),
|
|
_entry(
|
|
"2023-03-31",
|
|
"2023-01-01",
|
|
form="10-Q",
|
|
fp="Q1",
|
|
val=200.0 * m,
|
|
),
|
|
_entry(
|
|
"2023-04-01",
|
|
"2023-01-02",
|
|
form="10-Q",
|
|
fp="Q1",
|
|
filed="2024-05-10",
|
|
val=200.0 * m,
|
|
),
|
|
_entry(
|
|
"2023-06-30",
|
|
"2023-04-01",
|
|
form="10-Q",
|
|
fp="Q2",
|
|
val=250.0 * m,
|
|
),
|
|
_entry(
|
|
"2023-09-30",
|
|
"2023-07-01",
|
|
form="10-Q",
|
|
fp="Q3",
|
|
val=300.0 * m,
|
|
),
|
|
],
|
|
),
|
|
]
|
|
),
|
|
}
|
|
results = schema.extract_all(facts_json, frequency="quarterly")
|
|
income = results["income_statement"]
|
|
assert "2023-12-31" in income.dates
|
|
revenue = next(r for r in income.rows if r.tag == "total_revenue")
|
|
assert revenue.values["2023-12-31"] == 250.0 * m
|
|
|
|
|
|
class TestSchemaExtractAllErrors:
|
|
def test_quarterly_no_dates_generic_error(self, schema):
|
|
# A plain 10-K filer (not 20-F/40-F) with no interim data yields no
|
|
# quarterly filing dates and is not "annual-only" by form, so
|
|
# extract_all raises the generic no-quarterly OpenBBError (line 493).
|
|
from openbb_core.app.model.abstract.error import OpenBBError
|
|
|
|
facts_json = {
|
|
"entityName": "Domestic Annual Inc.",
|
|
"facts": _facts(
|
|
[
|
|
("us-gaap", "Assets", "USD", [_entry("2023-12-31", form="10-K")]),
|
|
(
|
|
"us-gaap",
|
|
"Revenues",
|
|
"USD",
|
|
[_entry("2023-12-31", "2023-01-01", form="10-K")],
|
|
),
|
|
]
|
|
),
|
|
}
|
|
with pytest.raises(OpenBBError, match="No quarterly filing dates"):
|
|
schema.extract_all(facts_json, frequency="quarterly")
|
|
|
|
|
|
class TestSchemaMergeFactsMetadata:
|
|
def test_merge_carries_label_and_description(self, schema):
|
|
# merge_facts copies label/description from the first source that has
|
|
# them (line 617-621).
|
|
a = {
|
|
"facts": {
|
|
"us-gaap": {
|
|
"Assets": {
|
|
"label": "Total Assets",
|
|
"description": "Sum of assets.",
|
|
"units": {"USD": [_entry("2022-12-31", val=1.0)]},
|
|
}
|
|
}
|
|
}
|
|
}
|
|
b = {
|
|
"facts": {
|
|
"us-gaap": {
|
|
"Assets": {
|
|
"label": "Ignored",
|
|
"units": {"USD": [_entry("2023-12-31", val=2.0)]},
|
|
}
|
|
}
|
|
}
|
|
}
|
|
merged = schema.merge_facts(a, b)
|
|
assert merged["us-gaap"]["Assets"]["label"] == "Total Assets"
|
|
assert merged["us-gaap"]["Assets"]["description"] == "Sum of assets."
|
|
assert len(merged["us-gaap"]["Assets"]["units"]["USD"]) == 2
|