OpenBB/openbb_platform/providers/sec/tests/test_imputation.py

"""Unit tests for ``openbb_sec.utils.statement_schema._imputation``.

These drive the imputation/verification engine directly with crafted
synthetic statements -- the engine is pure, so no network is involved.
They cover the source-formatting and multi-pass solver helpers, the
hierarchical roll-up/plug articulation, the per-statement ``impute()``
paths for income statement, balance sheet and cash flow, the equity-method
and ProfitLoss pretax corrections, the fiscal-year-end reconciliation, and
the many fact-based reconciliation fallbacks plus the final identity
enforcement and pending-diagnostic resolution.

Tests only -- no source under ``openbb_sec/`` is modified.
"""

# flake8: noqa: D101,D102,D103,D403

from unittest.mock import patch

import pytest

from openbb_sec.utils.statement_schema._imputation import (
    _apply_hierarchical_articulation,
    _format_impute_source,
    _formula_terms,
    _net_change_from_balances,
    _resolve_sign_flips,
    _run_imputation_passes,
    impute,
    reconcile_fiscal_year_ends,
)
from openbb_sec.utils.statement_schema._types import (
    RowDef,
    RowResult,
    StatementResult,
)

_M = 1_000_000
_D = "2023-12-31"


# ---------------------------------------------------------------------------
# Builders
# ---------------------------------------------------------------------------


def _rr(
    tag,
    values=None,
    *,
    parent=None,
    factor="+",
    balance="",
    sequence=1,
    period_type="duration",
    unit="monetary",
    sources=None,
    label=None,
):
    return RowResult(
        tag=tag,
        label=label or tag.replace("_", " ").title(),
        description="",
        parent=parent,
        sequence=sequence,
        factor=factor,
        balance=balance,
        unit=unit,
        period_type=period_type,
        values=dict(values or {}),
        sources=dict(sources or {}),
    )


def _rd(
    tag,
    *,
    xbrl=(),
    unit="monetary",
    period_type="duration",
    parent=None,
    factor="+",
    balance="",
    sequence=1,
    label=None,
):
    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),
    )


def _by_tag(rows, tag):
    for r in rows:
        if r.tag == tag:
            return r
    return None


def _dur(end, start, val, *, form="10-K", filed="2024-02-15"):
    return {"end": end, "start": start, "val": val, "form": form, "filed": filed}


def _inst(end, val, *, form="10-K", filed="2024-02-15"):
    return {"end": end, "val": val, "form": form, "filed": filed}


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


class TestFormatImputeSource:
    def test_single_positive_no_leading_plus(self):
        assert _format_impute_source("p", [("a", 1)]) == "p: a"

    def test_single_negative_leading_minus(self):
        assert _format_impute_source("p", [("a", -1)]) == "p: -a"


class TestRunImputationPasses:
    def test_source_tag_absent_from_index(self):
        # A rule whose source tag is not present at all -> all_present False, no change.
        d = _D
        rows = [_rr("target", {})]
        idx = {r.tag: i for i, r in enumerate(rows)}
        rules = [("target", [("missing_src", 1)])]
        assert _run_imputation_passes(rows, rules, idx, {d}) is False

    def test_target_already_has_value_is_skipped(self):
        d = _D
        rows = [_rr("target", {d: 5.0}), _rr("a", {d: 9.0})]
        idx = {r.tag: i for i, r in enumerate(rows)}
        rules = [("target", [("a", 1)])]
        # target already populated -> untouched, returns False.
        assert _run_imputation_passes(rows, rules, idx, {d}) is False
        assert rows[0].values[d] == 5.0

    def test_target_tag_not_in_index_skipped(self):
        d = _D
        rows = [_rr("a", {d: 9.0})]
        idx = {r.tag: i for i, r in enumerate(rows)}
        rules = [("nonexistent_target", [("a", 1)])]
        assert _run_imputation_passes(rows, rules, idx, {d}) is False

    def test_rolled_up_ancestor_source_is_skipped(self):
        d = _D
        rows = [
            _rr("parent", {d: 30.0}, sources={d: "imputed-rollup: a(+)"}),
            _rr("a", {d: 30.0}, parent="parent"),
            _rr("b", {}, parent="parent"),
        ]
        idx = {r.tag: i for i, r in enumerate(rows)}
        rules = [("b", [("parent", 1), ("a", -1)])]
        assert _run_imputation_passes(rows, rules, idx, {d}) is False
        assert d not in rows[2].values

    def test_tagged_ancestor_source_is_used(self):
        d = _D
        rows = [
            _rr("parent", {d: 30.0}, sources={d: "us-gaap:Parent"}),
            _rr("a", {d: 20.0}, parent="parent"),
            _rr("b", {}, parent="parent"),
        ]
        idx = {r.tag: i for i, r in enumerate(rows)}
        rules = [("b", [("parent", 1), ("a", -1)])]
        assert _run_imputation_passes(rows, rules, idx, {d}) is True
        assert rows[2].values[d] == 10.0

    @pytest.mark.parametrize(
        ("pretax", "costs_source", "expected"),
        [
            (100, "us-gaap:CostsAndExpenses", None),
            (80, "us-gaap:CostsAndExpenses", 300 * _M),
            (100, "imputed: total_revenue - total_operating_income", 300 * _M),
        ],
    )
    def test_single_step_costs_total_is_no_input(self, pretax, costs_source, expected):
        d = _D
        rows = [
            _rr("total_revenue", {d: 1000 * _M}, sources={d: "us-gaap:Revenues"}),
            _rr("costs_and_expenses", {d: 900 * _M}, sources={d: costs_source}),
            _rr("total_pretax_income", {d: pretax * _M}, sources={d: "us-gaap:P"}),
            _rr("total_cost_of_revenue", {d: 600 * _M}, sources={d: "us-gaap:C"}),
            _rr("total_operating_expenses", {}),
        ]
        rules = [
            (
                "total_operating_expenses",
                [("costs_and_expenses", 1), ("total_cost_of_revenue", -1)],
            )
        ]
        _run_imputation_passes(rows, rules, {r.tag: i for i, r in enumerate(rows)}, {d})
        assert _by_tag(rows, "total_operating_expenses").values.get(d) == expected

    @pytest.mark.parametrize(("operating", "expected"), [(500, None), (400, 100 * _M)])
    def test_zero_gross_line_not_derived(self, operating, expected):
        d = _D
        rows = [
            _rr("costs_and_expenses", {d: 500 * _M}, sources={d: "us-gaap:C"}),
            _rr(
                "total_operating_expenses",
                {d: operating * _M},
                sources={d: "us-gaap:O"},
            ),
            _rr("total_cost_of_revenue", {}),
        ]
        rules = [
            (
                "total_cost_of_revenue",
                [("costs_and_expenses", 1), ("total_operating_expenses", -1)],
            )
        ]
        _run_imputation_passes(rows, rules, {r.tag: i for i, r in enumerate(rows)}, {d})
        assert _by_tag(rows, "total_cost_of_revenue").values.get(d) == expected

    @pytest.mark.parametrize(("net_income", "expected"), [(70, None), (None, 300)])
    def test_single_step_with_derived_pretax(self, net_income, expected):
        d = _D
        rows = [
            _rr("total_revenue", {d: 1000 * _M}, sources={d: "us-gaap:Revenues"}),
            _rr(
                "costs_and_expenses",
                {d: 900 * _M},
                sources={d: "us-gaap:CostsAndExpenses"},
            ),
            _rr("total_pretax_income", {}),
            _rr(
                "net_income_continuing",
                {} if net_income is None else {d: net_income * _M},
                sources={} if net_income is None else {d: "us-gaap:N"},
            ),
            _rr("income_tax_expense", {d: 30 * _M}, sources={d: "us-gaap:T"}),
            _rr("total_cost_of_revenue", {d: 600 * _M}, sources={d: "us-gaap:C"}),
            _rr("total_operating_expenses", {}),
        ]
        rules = [
            (
                "total_operating_expenses",
                [("costs_and_expenses", 1), ("total_cost_of_revenue", -1)],
            ),
            (
                "total_pretax_income",
                [("total_revenue", 1), ("costs_and_expenses", -1)],
            ),
            (
                "total_pretax_income",
                [("net_income_continuing", 1), ("income_tax_expense", 1)],
            ),
        ]
        _run_imputation_passes(rows, rules, {r.tag: i for i, r in enumerate(rows)}, {d})
        opex = _by_tag(rows, "total_operating_expenses").values.get(d)
        assert opex == (None if expected is None else expected * _M)
        assert _by_tag(rows, "total_pretax_income").values[d] == 100 * _M

    def test_ancestor_walk_stops_on_parent_cycle(self):
        d = _D
        rows = [
            _rr("x", {d: 5.0}, parent="y"),
            _rr("y", {}, parent="x"),
            _rr("z", {d: 7.0}, sources={d: "imputed-rollup: q(+)"}),
        ]
        idx = {r.tag: i for i, r in enumerate(rows)}
        rules = [("y", [("z", 1)])]
        assert _run_imputation_passes(rows, rules, idx, {d}) is True
        assert rows[1].values[d] == 7.0


class TestApplyHierarchicalArticulation:
    def test_no_child_values_leaves_parent_untouched(self):
        d = _D
        rows = [
            _rr(
                "total_assets", {}, period_type="instant", balance="debit", sequence=10
            ),
            _rr(
                "cash",
                {},
                parent="total_assets",
                balance="debit",
                sequence=1,
                period_type="instant",
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        # No child had a value -> parent stays empty, no plug created.
        assert d not in _by_tag(rows, "total_assets").values
        assert _by_tag(rows, "other_assets") is None

    def test_tagged_other_line_replaced_by_remainder(self):
        d = _D
        rows = [
            _rr(
                "total_assets",
                {d: 200.0 * _M},
                period_type="instant",
                balance="debit",
                sequence=10,
            ),
            _rr(
                "cash",
                {d: 100.0 * _M},
                parent="total_assets",
                balance="debit",
                sequence=1,
                period_type="instant",
            ),
            _rr(
                "other_assets",
                {d: 7.0 * _M},
                parent="total_assets",
                balance="debit",
                sequence=2,
                period_type="instant",
                sources={d: "us-gaap:OtherAssets"},
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        oa = _by_tag(rows, "other_assets")
        assert oa.values[d] == 100.0 * _M
        assert oa.sources[d].startswith("imputed-plug")

    def test_tagged_other_line_kept_when_articulating(self):
        d = _D
        rows = [
            _rr("total_assets", {d: 200.0 * _M}, period_type="instant", sequence=10),
            _rr(
                "cash",
                {d: 100.0 * _M},
                parent="total_assets",
                sequence=1,
                period_type="instant",
            ),
            _rr(
                "other_assets",
                {d: 100.0 * _M},
                parent="total_assets",
                sequence=2,
                period_type="instant",
                sources={d: "us-gaap:OtherAssets"},
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "other_assets").sources[d] == "us-gaap:OtherAssets"

    def test_mapped_other_line_holds_remainder(self):
        d = _D
        rows = [
            _rr("total_other_income", {d: 50.0 * _M}, sequence=10),
            _rr(
                "total_interest_income",
                {d: 20.0 * _M},
                parent="total_other_income",
                sequence=1,
            ),
            _rr(
                "other_income",
                {d: 5.0 * _M},
                parent="total_other_income",
                sequence=2,
                sources={d: "us-gaap:OtherNonoperatingIncomeExpense"},
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        oi = _by_tag(rows, "other_income")
        assert oi.values[d] == 30.0 * _M
        assert oi.sources[d].startswith("imputed-plug")
        assert _by_tag(rows, "other_other_income") is None

    def test_imputed_plug_child_excluded_then_replug(self):
        # An other_* child already carrying an imputed-plug is excluded from the
        # children sum and re-plugged to the fresh remainder.
        d = _D
        rows = [
            _rr(
                "total_assets",
                {d: 200.0 * _M},
                period_type="instant",
                balance="debit",
                sequence=10,
            ),
            _rr(
                "cash",
                {d: 120.0 * _M},
                parent="total_assets",
                balance="debit",
                sequence=1,
                period_type="instant",
            ),
            _rr(
                "other_assets",
                {d: 999.0 * _M},
                parent="total_assets",
                balance="debit",
                sequence=2,
                period_type="instant",
                sources={d: "imputed-plug: stale"},
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        oa = _by_tag(rows, "other_assets")
        # remainder = 200 - 120 = 80 (stale plug excluded from the sum)
        assert oa.values[d] == 80.0 * _M
        assert "imputed-plug" in oa.sources[d]

    def test_rollup_requires_required_child_value(self):
        d = _D
        rows = [
            _rr("total_gross_profit", {}, sequence=10),
            _rr(
                "total_revenue",
                {d: 100.0 * _M},
                parent="total_gross_profit",
                sequence=1,
            ),
            _rr(
                "total_cost_of_revenue",
                {},
                parent="total_gross_profit",
                factor="-",
                sequence=2,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert d not in _by_tag(rows, "total_gross_profit").values

    def test_rollup_requirement_ignored_when_not_in_tree(self):
        d = _D
        rows = [
            _rr("total_liabilities", {}, period_type="instant", sequence=10),
            _rr(
                "total_current_liabilities",
                {d: 40.0 * _M},
                parent="total_liabilities",
                period_type="instant",
                sequence=1,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "total_liabilities").values[d] == 40.0 * _M

    def test_empty_rollup_requirement_never_rolls_up(self):
        d = _D
        rows = [
            _rr("comprehensive_income_parent", {}, sequence=10),
            _rr(
                "other_comprehensive_income",
                {d: 5.0 * _M},
                parent="comprehensive_income_parent",
                sequence=1,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert d not in _by_tag(rows, "comprehensive_income_parent").values

    def test_rollup_includes_designated_other_line(self):
        d = _D
        rows = [
            _rr("total_other_income", {}, sequence=10),
            _rr(
                "total_interest_income",
                {d: 20.0 * _M},
                parent="total_other_income",
                sequence=1,
            ),
            _rr(
                "other_income",
                {d: 5.0 * _M},
                parent="total_other_income",
                sequence=2,
                sources={d: "us-gaap:OtherNonoperatingIncomeExpense"},
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        total_other = _by_tag(rows, "total_other_income")
        assert total_other.values[d] == 25.0 * _M
        assert total_other.sources[d] == (
            "imputed-rollup: total_interest_income(+) + other_income(+)"
        )

    def test_rollup_from_designated_other_line_alone(self):
        d = _D
        rows = [
            _rr("total_other_income", {}, sequence=10),
            _rr(
                "other_income",
                {d: 5.0 * _M},
                parent="total_other_income",
                sequence=2,
                sources={d: "us-gaap:OtherNonoperatingIncomeExpense"},
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        total_other = _by_tag(rows, "total_other_income")
        assert total_other.values[d] == 5.0 * _M
        assert total_other.sources[d] == "imputed-rollup: other_income(+)"

    def test_contained_component_gets_period_factor_zero(self):
        d = _D
        rows = [
            _rr("total_x", {d: 100.0 * _M}, sequence=10),
            _rr("a", {d: 60.0 * _M}, parent="total_x", sequence=1),
            _rr("b", {d: 40.0 * _M}, parent="total_x", sequence=2),
            _rr("c", {d: 30.0 * _M}, parent="total_x", sequence=3),
        ]
        _apply_hierarchical_articulation(rows, {d})
        c = _by_tag(rows, "c")
        assert c.date_factors == {d: "0"}
        assert c.factor_on(d) == "0"
        assert c.values[d] == 30.0 * _M
        assert _by_tag(rows, "other_x") is None

    def test_contained_pair_gets_period_factor_zero(self):
        d = _D
        rows = [
            _rr("total_x", {d: 100.0 * _M}, sequence=10),
            _rr("a", {d: 70.0 * _M}, parent="total_x", sequence=1),
            _rr("b", {d: 30.0 * _M}, parent="total_x", sequence=2),
            _rr("c", {d: 12.0 * _M}, parent="total_x", sequence=3),
            _rr("e", {d: 8.0 * _M}, parent="total_x", sequence=4),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "c").factor_on(d) == "0"
        assert _by_tag(rows, "e").factor_on(d) == "0"
        assert _by_tag(rows, "a").factor_on(d) == "+"
        assert _by_tag(rows, "other_x") is None

    def test_ambiguous_containment_goes_to_remainder(self):
        d = _D
        rows = [
            _rr("total_x", {d: 100.0 * _M}, sequence=10),
            _rr("a", {d: 80.0 * _M}, parent="total_x", sequence=1),
            _rr("b", {d: 20.0 * _M}, parent="total_x", sequence=2),
            _rr("c", {d: 20.0 * _M}, parent="total_x", sequence=3),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "b").date_factors == {}
        assert _by_tag(rows, "c").date_factors == {}
        other = _by_tag(rows, "other_x")
        assert other.values[d] == -20.0 * _M
        assert other.sources[d].startswith("imputed-plug: total_x - ")

    def test_contained_single_equal_to_other_line_left_to_other_line(self):
        d = _D
        rows = [
            _rr("total_x", {d: 100.0 * _M}, sequence=10),
            _rr("a", {d: 60.0 * _M}, parent="total_x", sequence=1),
            _rr("b", {d: 40.0 * _M}, parent="total_x", sequence=2),
            _rr(
                "other_x",
                {d: 40.0 * _M},
                parent="total_x",
                sequence=3,
                sources={d: "us-gaap:OtherX"},
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "b").date_factors == {}
        assert d not in _by_tag(rows, "other_x").values

    def test_explicit_line_inside_tagged_other_contained(self):
        d = _D
        rows = [
            _rr(
                "total_noncurrent_assets",
                {d: 100.0 * _M},
                period_type="instant",
                sequence=10,
            ),
            _rr(
                "net_ppe",
                {d: 60.0 * _M},
                parent="total_noncurrent_assets",
                period_type="instant",
                sequence=1,
            ),
            _rr(
                "operating_lease_right_of_use_asset",
                {d: 10.0 * _M},
                parent="total_noncurrent_assets",
                period_type="instant",
                sequence=2,
            ),
            _rr(
                "other_noncurrent_assets",
                {d: 40.0 * _M},
                parent="total_noncurrent_assets",
                period_type="instant",
                sequence=3,
                sources={d: "us-gaap:OtherAssetsNoncurrent"},
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "operating_lease_right_of_use_asset").factor_on(d) == "0"
        other = _by_tag(rows, "other_noncurrent_assets")
        assert other.values[d] == 40.0 * _M
        assert other.sources[d] == "us-gaap:OtherAssetsNoncurrent"

    def test_component_without_room_not_contained(self):
        d = _D
        rows = [
            _rr("total_x", {d: -103.0 * _M}, sequence=10),
            _rr("a", {d: -60.0 * _M}, parent="total_x", sequence=1),
            _rr("b", {d: -40.0 * _M}, parent="total_x", sequence=2),
            _rr("c", {d: 50.0 * _M}, parent="total_x", sequence=3),
            _rr(
                "other_x",
                {d: -3.0 * _M},
                parent="total_x",
                sequence=4,
                sources={d: "us-gaap:OtherX"},
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "c").date_factors == {}
        other = _by_tag(rows, "other_x")
        assert other.values[d] == -53.0 * _M
        assert other.sources[d].startswith("imputed-plug")

    def test_scope_variant_contained_without_room(self):
        d = _D
        rows = [
            _rr("total_pretax_income", {d: 100.0 * _M}, sequence=10),
            _rr(
                "total_operating_income",
                {d: 100.0 * _M},
                parent="total_pretax_income",
                sequence=1,
            ),
            _rr(
                "equity_method_investments",
                {d: 150.0 * _M},
                parent="total_pretax_income",
                sequence=2,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "equity_method_investments").factor_on(d) == "0"

    def test_net_income_line_holds_no_component(self):
        d = _D
        rows = [
            _rr(
                "net_cash_from_continuing_operating_activities",
                {d: 100.0 * _M},
                sequence=10,
            ),
            _rr(
                "net_income_continuing",
                {d: 100.0 * _M},
                parent="net_cash_from_continuing_operating_activities",
                sequence=1,
            ),
            _rr(
                "stock_based_compensation",
                {d: 30.0 * _M},
                parent="net_cash_from_continuing_operating_activities",
                sequence=2,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "stock_based_compensation").date_factors == {}
        other = _by_tag(rows, "other_operating_activities") or _by_tag(
            rows, "other_net_cash_from_continuing_operating_activities"
        )
        assert other.values[d] == -30.0 * _M

    def test_mixed_sign_pair_not_contained(self):
        d = _D
        rows = [
            _rr("total_x", {d: 100.0 * _M}, sequence=10),
            _rr("a", {d: 100.0 * _M}, parent="total_x", sequence=1),
            _rr("b", {d: 193.0 * _M}, parent="total_x", sequence=2),
            _rr("c", {d: -124.0 * _M}, parent="total_x", sequence=3),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "b").date_factors == {}
        assert _by_tag(rows, "c").date_factors == {}
        assert _by_tag(rows, "other_x").values[d] == -69.0 * _M

    def test_component_outside_reported_section_total_relocated(self):
        d = _D

        def inst(tag, value, parent, seq):
            return _rr(
                tag,
                {} if value is None else {d: value * _M},
                parent=parent,
                period_type="instant",
                sequence=seq,
            )

        rows = [
            inst("total_liabilities_and_equity", 6036.0, None, 30),
            inst("total_liabilities", None, "total_liabilities_and_equity", 20),
            inst("total_current_liabilities", 440.0, "total_liabilities", 1),
            inst("total_noncurrent_liabilities", 1528.0, "total_liabilities", 10),
            inst("long_term_debt", 2131.0, "total_noncurrent_liabilities", 2),
            inst(
                "asset_retirement_and_litigation_obligation",
                108.0,
                "total_noncurrent_liabilities",
                3,
            ),
            inst(
                "noncurrent_deferred_tax_liabilities",
                513.0,
                "total_noncurrent_liabilities",
                4,
            ),
            inst(
                "total_equity_and_noncontrolling_interests",
                1937.0,
                "total_liabilities_and_equity",
                21,
            ),
            inst("temporary_equity", None, "total_liabilities_and_equity", 22),
        ]
        _apply_hierarchical_articulation(rows, {d})
        tncl = _by_tag(rows, "total_noncurrent_liabilities")
        assert tncl.values[d] == 3659.0 * _M
        assert tncl.sources[d] == (
            "corrected: total_noncurrent_liabilities + long_term_debt"
        )
        assert _by_tag(rows, "other_noncurrent_liabilities").values[d] == 907.0 * _M
        assert _by_tag(rows, "total_liabilities").values[d] == 4099.0 * _M
        assert d not in _by_tag(rows, "temporary_equity").values

    def _inst(self, tag, value, parent=None, seq=1, source=None):
        d = _D
        return _rr(
            tag,
            {} if value is None else {d: value * _M},
            parent=parent,
            period_type="instant",
            sequence=seq,
            sources={} if value is None or source is None else {d: source},
        )

    def test_same_fact_in_two_rows_contained_once(self):
        d = _D
        fact = "us-gaap:PropertyPlantAndEquipmentNet"
        rows = [
            self._inst("total_assets", 100.0, seq=20),
            self._inst("cash_and_equivalents", 70.0, "total_assets", 1),
            self._inst("net_ppe", 30.0, "total_assets", 5, fact),
            self._inst("net_premises_and_equipment", 30.0, "total_assets", 6, fact),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "net_ppe").date_factors == {}
        assert _by_tag(rows, "net_premises_and_equipment").factor_on(d) == "0"
        assert _by_tag(rows, "other_assets") is None

    def test_equal_balances_over_total_contained_once(self):
        d = _D
        rows = [
            self._inst("total_current_assets", 18452.0, seq=20),
            self._inst("cash_and_equivalents", 3818.0, "total_current_assets", 1),
            self._inst("short_term_investments", 90.0, "total_current_assets", 2),
            self._inst("accounts_receivable", 6673.0, "total_current_assets", 3),
            self._inst("net_inventory", 3886.0, "total_current_assets", 4),
            self._inst(
                "prepaid_expenses",
                2531.0,
                "total_current_assets",
                5,
                "us-gaap:PrepaidExpenseCurrent",
            ),
            self._inst(
                "other_current_nonoperating_assets",
                2531.0,
                "total_current_assets",
                8,
                "us-gaap:PrepaidExpenseAndOtherAssetsCurrent",
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "prepaid_expenses").date_factors == {}
        assert _by_tag(rows, "other_current_nonoperating_assets").factor_on(d) == "0"
        assert _by_tag(rows, "other_current_assets").values[d] == 1454.0 * _M

    def test_nested_group_leaving_smallest_remainder_contained(self):
        d = _D
        rows = [
            self._inst("total_current_liabilities", 30011.0, seq=20),
            self._inst("accounts_payable", 11260.0, "total_current_liabilities", 1),
            self._inst("accrued_expenses", 9054.0, "total_current_liabilities", 2),
            self._inst("short_term_debt", 6183.0, "total_current_liabilities", 3),
            self._inst(
                "current_portion_of_long_term_debt",
                3388.0,
                "total_current_liabilities",
                4,
            ),
            self._inst(
                "current_employee_benefit_liabilities",
                1623.0,
                "total_current_liabilities",
                5,
            ),
            self._inst("other_taxes_payable", 341.0, "total_current_liabilities", 6),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "current_employee_benefit_liabilities").factor_on(d) == "0"
        assert _by_tag(rows, "other_taxes_payable").factor_on(d) == "0"
        assert _by_tag(rows, "current_portion_of_long_term_debt").date_factors == {}
        assert _by_tag(rows, "other_current_liabilities").values[d] == 126.0 * _M

    def test_total_equity_reconciled_with_common_equity_rollup(self):
        d = _D
        rows = [
            self._inst("total_equity", 100.0, seq=20),
            self._inst("total_common_equity", None, "total_equity", 10),
            self._inst("common_equity", 10.0, "total_common_equity", 1),
            self._inst("additional_paid_in_capital", 50.0, "total_common_equity", 2),
            self._inst("retained_earnings", 30.0, "total_common_equity", 3),
            self._inst("other_equity", None, "total_common_equity", 4),
        ]
        _apply_hierarchical_articulation(rows, {d})
        tce = _by_tag(rows, "total_common_equity")
        assert tce.values[d] == 100.0 * _M
        assert tce.sources[d].startswith("imputed-plug: total_equity - ")
        other = _by_tag(rows, "other_equity")
        assert other.values[d] == 10.0 * _M
        assert other.sources[d].startswith("imputed-plug: total_common_equity - ")

    def test_contra_only_components_create_no_remainder(self):
        d = _D
        rows = [
            self._inst("net_ppe", 24675.0, seq=20),
            self._inst("gross_ppe", None, "net_ppe", 1),
            _rr(
                "accumulated_depreciation",
                {d: 12560.0 * _M},
                parent="net_ppe",
                factor="-",
                period_type="instant",
                sequence=2,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "other_net_ppe") is None
        assert _by_tag(rows, "net_ppe").values[d] == 24675.0 * _M

    def test_scope_variant_pair_of_mixed_signs_contained(self):
        d = _D
        rows = [
            _rr("total_x", {d: 47.0 * _M}, sequence=10),
            _rr("a", {d: 1552.0 * _M}, parent="total_x", sequence=1),
            _rr(
                "net_cash_from_discontinued_operating_activities",
                {d: -7.0 * _M},
                parent="total_x",
                sequence=2,
            ),
            _rr("b", {d: -663.0 * _M}, parent="total_x", sequence=3),
            _rr(
                "net_cash_from_discontinued_investing_activities",
                {d: 50.0 * _M},
                parent="total_x",
                sequence=4,
            ),
            _rr("c", {d: -842.0 * _M}, parent="total_x", sequence=5),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert (
            _by_tag(rows, "net_cash_from_discontinued_operating_activities").factor_on(
                d
            )
            == "0"
        )
        assert (
            _by_tag(rows, "net_cash_from_discontinued_investing_activities").factor_on(
                d
            )
            == "0"
        )
        assert _by_tag(rows, "other_x") is None

    def test_imputed_parent_is_not_containment_evidence(self):
        d = _D
        rows = [
            _rr(
                "total_x",
                {d: 100.0 * _M},
                sequence=10,
                sources={d: "imputed: total_y - total_z"},
            ),
            _rr("a", {d: 60.0 * _M}, parent="total_x", sequence=1),
            _rr("b", {d: 40.0 * _M}, parent="total_x", sequence=2),
            _rr("c", {d: 30.0 * _M}, parent="total_x", sequence=3),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "c").date_factors == {}
        assert _by_tag(rows, "other_x").values[d] == -30.0 * _M

    def test_every_same_fact_duplicate_contained(self):
        d = _D
        ppe = "us-gaap:PropertyPlantAndEquipmentNet"
        receivables = "us-gaap:AccountsReceivableNet"
        rows = [
            self._inst("total_assets", 65.0, seq=20),
            self._inst("cash_and_equivalents", 30.0, "total_assets", 1),
            self._inst("net_ppe", 10.0, "total_assets", 5, ppe),
            self._inst("net_premises_and_equipment", 10.0, "total_assets", 6, ppe),
            self._inst("accounts_receivable", 25.0, "total_assets", 7, receivables),
            self._inst(
                "customer_and_other_receivables", 25.0, "total_assets", 8, receivables
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "net_premises_and_equipment").factor_on(d) == "0"
        assert _by_tag(rows, "customer_and_other_receivables").factor_on(d) == "0"
        assert _by_tag(rows, "net_ppe").date_factors == {}
        assert _by_tag(rows, "accounts_receivable").date_factors == {}

    def test_contained_component_found_under_rolled_up_other_line(self):
        d = _D
        rows = [
            _rr("total_equity", {d: 100.0 * _M}, period_type="instant", sequence=10),
            _rr(
                "total_preferred_equity",
                {d: 10.0 * _M},
                parent="total_equity",
                period_type="instant",
                sequence=1,
            ),
            _rr(
                "total_common_equity",
                {},
                parent="total_equity",
                period_type="instant",
                sequence=2,
            ),
            _rr(
                "retained_earnings",
                {d: 90.0 * _M},
                parent="total_common_equity",
                period_type="instant",
                sequence=3,
            ),
            _rr(
                "preferred_in_common",
                {d: 10.0 * _M},
                parent="total_common_equity",
                period_type="instant",
                sequence=4,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "preferred_in_common").factor_on(d) == "0"
        assert _by_tag(rows, "total_common_equity").values[d] == 90.0 * _M

    def test_presented_memo_promoted(self):
        d = _D
        rows = [
            _rr("net_cash_from_operating_activities", {d: 100.0 * _M}, sequence=10),
            _rr(
                "net_cash_from_continuing_operating_activities",
                {d: 70.0 * _M},
                parent="net_cash_from_operating_activities",
                sequence=1,
            ),
            _rr(
                "net_cash_from_discontinued_operations",
                {d: 30.0 * _M},
                parent="net_cash_from_operating_activities",
                factor="0",
                sequence=2,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        memo = _by_tag(rows, "net_cash_from_discontinued_operations")
        assert memo.factor_on(d) == "+"
        assert memo.factor == "0"
        assert _by_tag(rows, "other_net_cash_from_operating_activities") is None

    def test_incomplete_rollup_absorbs_section_difference(self):
        d = _D
        rows = [
            _rr(
                "total_liabilities_and_equity",
                {d: 300.0 * _M},
                period_type="instant",
                sequence=20,
            ),
            _rr(
                "total_liabilities",
                {},
                parent="total_liabilities_and_equity",
                period_type="instant",
                sequence=10,
            ),
            _rr(
                "total_current_liabilities",
                {d: 100.0 * _M},
                parent="total_liabilities",
                period_type="instant",
                sequence=1,
            ),
            _rr(
                "long_term_debt",
                {d: 50.0 * _M},
                parent="total_liabilities",
                period_type="instant",
                sequence=2,
            ),
            _rr(
                "total_equity_and_noncontrolling_interests",
                {d: 100.0 * _M},
                parent="total_liabilities_and_equity",
                period_type="instant",
                sequence=11,
            ),
            _rr(
                "temporary_equity",
                {},
                parent="total_liabilities_and_equity",
                period_type="instant",
                sequence=12,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        tl = _by_tag(rows, "total_liabilities")
        assert tl.values[d] == 200.0 * _M
        assert tl.sources[d].startswith("imputed-plug: total_liabilities_and_equity - ")
        other = _by_tag(rows, "other_liabilities")
        assert other.values[d] == 50.0 * _M
        assert d not in _by_tag(rows, "temporary_equity").values

    def test_nonnegative_line_negative_remainder_not_held(self):
        d = _D
        rows = [
            _rr(
                "total_liabilities_and_equity",
                {d: 300.0 * _M},
                period_type="instant",
                sequence=20,
            ),
            _rr(
                "total_liabilities",
                {d: 220.0 * _M},
                parent="total_liabilities_and_equity",
                period_type="instant",
                sequence=10,
            ),
            _rr(
                "total_equity_and_noncontrolling_interests",
                {d: 100.0 * _M},
                parent="total_liabilities_and_equity",
                period_type="instant",
                sequence=11,
            ),
            _rr(
                "temporary_equity",
                {},
                parent="total_liabilities_and_equity",
                period_type="instant",
                sequence=12,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert d not in _by_tag(rows, "temporary_equity").values
        assert _by_tag(rows, "other_liabilities_and_equity") is None

    def test_nonnegative_line_positive_remainder_held(self):
        d = _D
        rows = [
            _rr(
                "total_liabilities_and_equity",
                {d: 300.0 * _M},
                period_type="instant",
                sequence=20,
            ),
            _rr(
                "total_liabilities",
                {d: 180.0 * _M},
                parent="total_liabilities_and_equity",
                period_type="instant",
                sequence=10,
            ),
            _rr(
                "total_equity_and_noncontrolling_interests",
                {d: 100.0 * _M},
                parent="total_liabilities_and_equity",
                period_type="instant",
                sequence=11,
            ),
            _rr(
                "temporary_equity",
                {},
                parent="total_liabilities_and_equity",
                period_type="instant",
                sequence=12,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "temporary_equity").values[d] == 20.0 * _M
        assert _by_tag(rows, "other_liabilities_and_equity") is None

    def test_negative_noncurrent_remainder_not_held(self):
        d = _D
        rows = [
            _rr("total_assets", {d: 100.0 * _M}, period_type="instant", sequence=20),
            _rr(
                "total_current_assets",
                {d: 110.0 * _M},
                parent="total_assets",
                period_type="instant",
                sequence=1,
            ),
            _rr(
                "total_noncurrent_assets",
                {},
                parent="total_assets",
                period_type="instant",
                sequence=2,
            ),
            _rr(
                "other_assets",
                {},
                parent="total_assets",
                period_type="instant",
                sequence=3,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert d not in _by_tag(rows, "total_noncurrent_assets").values
        assert d not in _by_tag(rows, "other_assets").values

    def test_negative_gross_remainder_not_held(self):
        d = _D
        rows = [
            _rr("depreciation_and_amortization", {d: 100 * _M}, sequence=10),
            _rr(
                "depreciation_expense",
                {d: 80 * _M},
                parent="depreciation_and_amortization",
                sequence=1,
            ),
            _rr(
                "amortization_expense",
                {d: 30 * _M},
                parent="depreciation_and_amortization",
                sequence=2,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "other_depreciation_and_amortization") is None
        assert _by_tag(rows, "depreciation_and_amortization").values[d] == 100 * _M

    @pytest.mark.parametrize(("operating", "expected"), [(None, None), (50, 30 * _M)])
    def test_pretax_remainder_requires_operating_income(self, operating, expected):
        d = _D
        rows = [
            _rr("total_pretax_income", {d: 100 * _M}, sequence=10),
            _rr(
                "total_operating_income",
                {} if operating is None else {d: operating * _M},
                parent="total_pretax_income",
                sequence=1,
            ),
            _rr(
                "total_other_income",
                {},
                parent="total_pretax_income",
                sequence=2,
            ),
            _rr(
                "equity_method_investments",
                {d: 20 * _M},
                parent="total_pretax_income",
                sequence=3,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "total_other_income").values.get(d) == expected

    def test_all_zero_children_not_rolled_up(self):
        d = _D
        rows = [
            _rr("depreciation_and_amortization", {}, sequence=10),
            _rr("depreciation_expense", {d: 0}, parent="depreciation_and_amortization"),
            _rr("amortization_expense", {d: 0}, parent="depreciation_and_amortization"),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert d not in _by_tag(rows, "depreciation_and_amortization").values

    def test_contra_only_children_not_rolled_up(self):
        d = _D
        rows = [
            _rr("net_ppe", {}, period_type="instant", sequence=15),
            _rr("gross_ppe", {}, parent="net_ppe", period_type="instant", sequence=13),
            _rr(
                "accumulated_depreciation",
                {d: 531 * _M},
                parent="net_ppe",
                factor="-",
                period_type="instant",
                sequence=14,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert d not in _by_tag(rows, "net_ppe").values

    @pytest.mark.parametrize(("operating", "expected"), [(None, None), (500, 300)])
    def test_pretax_rollup_requires_operating_income(self, operating, expected):
        d = _D
        rows = [
            _rr("total_pretax_income", {}, sequence=10),
            _rr(
                "total_operating_income",
                {} if operating is None else {d: operating * _M},
                parent="total_pretax_income",
                sequence=1,
            ),
            _rr(
                "total_other_income",
                {d: -200 * _M},
                parent="total_pretax_income",
                sequence=2,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        pretax = _by_tag(rows, "total_pretax_income").values.get(d)
        assert pretax == (None if expected is None else expected * _M)

    @pytest.mark.parametrize(
        ("liabilities", "expected"), [(None, None), (600, 1000 * _M)]
    )
    def test_liabilities_and_equity_rollup_requires_liabilities(
        self, liabilities, expected
    ):
        d = _D
        rows = [
            _rr("total_liabilities_and_equity", {}, period_type="instant", sequence=73),
            _rr(
                "total_liabilities",
                {} if liabilities is None else {d: liabilities * _M},
                parent="total_liabilities_and_equity",
                period_type="instant",
                sequence=54,
            ),
            _rr(
                "total_equity_and_noncontrolling_interests",
                {d: 400 * _M},
                parent="total_liabilities_and_equity",
                period_type="instant",
                sequence=72,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "total_liabilities_and_equity").values.get(d) == expected

    def test_operating_totals_never_rolled_up(self):
        d = _D
        rows = [
            _rr("total_operating_income", {}, sequence=10),
            _rr(
                "total_gross_profit",
                {d: 100 * _M},
                parent="total_operating_income",
                sequence=1,
            ),
            _rr(
                "total_operating_expenses",
                {},
                parent="total_operating_income",
                factor="-",
                sequence=2,
            ),
            _rr("sga_expense", {d: 30 * _M}, parent="total_operating_expenses"),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert d not in _by_tag(rows, "total_operating_expenses").values
        assert d not in _by_tag(rows, "total_operating_income").values

    def test_integer_rollup_stays_integer(self):
        d = _D
        rows = [
            _rr("depreciation_and_amortization", {}, sequence=10),
            _rr(
                "depreciation_expense",
                {d: 80 * _M},
                parent="depreciation_and_amortization",
            ),
            _rr(
                "amortization_expense",
                {d: 30 * _M},
                parent="depreciation_and_amortization",
                factor="-",
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        value = _by_tag(rows, "depreciation_and_amortization").values[d]
        assert value == 50 * _M
        assert isinstance(value, int)

    @pytest.mark.parametrize(("liabilities", "expected"), [(None, None), (600, 1100)])
    def test_date_without_balance_sheet_holds_no_totals(self, liabilities, expected):
        d = _D

        def inst(tag, value, parent=None, seq=1):
            return _rr(
                tag,
                {} if value is None else {d: value * _M},
                parent=parent,
                period_type="instant",
                sequence=seq,
                sources={} if value is None else {d: f"us-gaap:{tag}"},
            )

        rows = [
            inst("total_assets", None, seq=10),
            inst("cash_and_equivalents", 100, "total_assets", 1),
            inst("total_liabilities_and_equity", None, seq=30),
            inst("total_liabilities", liabilities, "total_liabilities_and_equity", 20),
            inst(
                "total_equity_and_noncontrolling_interests",
                500,
                "total_liabilities_and_equity",
                25,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        le = _by_tag(rows, "total_liabilities_and_equity").values.get(d)
        assert le == (None if expected is None else expected * _M)
        if expected is None:
            assert d not in _by_tag(rows, "total_assets").values

    def test_unclassified_section_holds_no_totals(self):
        d = _D

        def inst(tag, value, parent=None, seq=1, source=None):
            return _rr(
                tag,
                {} if value is None else {d: value * _M},
                parent=parent,
                period_type="instant",
                sequence=seq,
                sources={} if value is None else {d: source or f"us-gaap:{tag}"},
            )

        rows = [
            inst("total_assets", 1000, seq=10),
            inst("total_current_assets", None, "total_assets", 1),
            inst("cash_and_equivalents", 300, "total_current_assets", 1),
            inst("accounts_receivable", 200, "total_current_assets", 2),
            inst(
                "other_current_assets",
                50,
                "total_current_assets",
                3,
                "imputed-plug: total_current_assets - (cash_and_equivalents(+))",
            ),
            inst("total_noncurrent_assets", None, "total_assets", 5),
            inst("goodwill", 400, "total_noncurrent_assets", 1),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert d not in _by_tag(rows, "total_current_assets").values
        assert d not in _by_tag(rows, "total_noncurrent_assets").values
        assert d not in _by_tag(rows, "other_current_assets").values
        assert _by_tag(rows, "other_noncurrent_assets") is None
        assert _by_tag(rows, "total_assets").values[d] == 1000 * _M

    def test_remainder_rolled_up_from_its_only_match_is_contained(self):
        d = _D

        def inst(tag, value, parent=None, seq=1):
            return _rr(
                tag,
                {} if value is None else {d: value * _M},
                parent=parent,
                period_type="instant",
                balance="credit",
                sequence=seq,
                sources={} if value is None else {d: f"us-gaap:{tag}"},
            )

        rows = [
            inst("total_liabilities_and_equity", 1000, seq=10),
            inst("total_liabilities", 700, "total_liabilities_and_equity", 1),
            inst(
                "total_equity_and_noncontrolling_interests",
                300,
                "total_liabilities_and_equity",
                2,
            ),
            inst("temporary_equity", None, "total_liabilities_and_equity", 3),
            inst("redeemable_noncontrolling_interest", 50, "temporary_equity", 1),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "redeemable_noncontrolling_interest").factor_on(d) == "0"
        assert d not in _by_tag(rows, "temporary_equity").values
        assert _by_tag(rows, "other_temporary_equity") is None

    def test_soft_total_fallback_rolled_up_from_children(self):
        d = _D
        rows = [
            _rr(
                "net_cash_from_operating_activities",
                {d: 999.0 * _M},
                sequence=10,
                sources={
                    d: "us-gaap:NetCashProvidedByUsedInOperatingActivities(fallback)"
                },
            ),
            _rr(
                "net_cash_from_continuing_operating_activities",
                {d: 80.0 * _M},
                parent="net_cash_from_operating_activities",
                sequence=1,
            ),
            _rr(
                "net_cash_from_discontinued_operating_activities",
                {d: 20.0 * _M},
                parent="net_cash_from_operating_activities",
                sequence=2,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        op = _by_tag(rows, "net_cash_from_operating_activities")
        assert op.values[d] == 100.0 * _M
        assert op.sources[d].startswith("imputed-rollup")

    def test_stale_other_plug_and_absorbable_child_go_to_remainder(self):
        d = _D
        rows = [
            _rr("total_operating_income", {d: 100.0 * _M}, sequence=20),
            _rr(
                "total_gross_profit",
                {d: 300.0 * _M},
                parent="total_operating_income",
                sequence=1,
            ),
            _rr(
                "total_operating_expenses",
                {d: 150.0 * _M},
                parent="total_operating_income",
                factor="-",
                sequence=2,
                sources={d: "imputed-plug: total_operating_income - (stale)"},
            ),
            _rr(
                "sga",
                {},
                parent="total_operating_expenses",
                sequence=3,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        opex = _by_tag(rows, "total_operating_expenses")
        assert opex.values[d] == 200.0 * _M
        assert opex.sources[d].startswith("imputed-plug: total_operating_income - ")

    def test_incomplete_rolled_up_other_line_is_absorber(self):
        d = _D

        def inst(tag, value, parent=None, seq=1, source=None):
            return _rr(
                tag,
                {} if value is None else {d: value * _M},
                parent=parent,
                period_type="instant",
                sequence=seq,
                sources={} if value is None else {d: source or f"us-gaap:{tag}"},
            )

        rows = [
            inst("total_liabilities", 500, seq=20),
            inst("total_current_liabilities", 200, "total_liabilities", 1),
            inst(
                "accrued_charges",
                5,
                "total_liabilities",
                2,
                "imputed-rollup: x(+) + y(+)",
            ),
            inst("total_noncurrent_liabilities", None, "total_liabilities", 3),
            inst("long_term_debt", 150, "total_noncurrent_liabilities", 1),
            inst(
                "noncurrent_deferred_tax_liabilities",
                50,
                "total_noncurrent_liabilities",
                2,
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        noncurrent = _by_tag(rows, "total_noncurrent_liabilities")
        assert noncurrent.values[d] == 295 * _M
        assert noncurrent.sources[d].startswith("imputed-plug: total_liabilities - ")
        assert _by_tag(rows, "other_noncurrent_liabilities").values[d] == 95 * _M

    @staticmethod
    def _le_rows(d, liabilities, current):
        def inst(tag, value, parent=None, seq=1):
            return _rr(
                tag,
                {} if value is None else {d: value * _M},
                parent=parent,
                period_type="instant",
                balance="credit",
                sequence=seq,
                sources={} if value is None else {d: f"us-gaap:{tag}"},
            )

        return [
            inst("total_liabilities_and_equity", 1000.0, seq=10),
            inst("total_liabilities", liabilities, "total_liabilities_and_equity", 2),
            inst("total_current_liabilities", current, "total_liabilities", 1),
            inst("temporary_equity", None, "total_liabilities_and_equity", 3),
            inst("redeemable_noncontrolling_interest", None, "temporary_equity", 1),
            inst(
                "total_equity_and_noncontrolling_interests",
                350.0,
                "total_liabilities_and_equity",
                4,
            ),
        ]

    def test_liabilities_rollup_absorbs_remainder_without_mezzanine(self):
        d = "2023-12-31"
        rows = self._le_rows(d, None, 300.0)
        _apply_hierarchical_articulation(rows, {d})
        tl = _by_tag(rows, "total_liabilities")
        assert tl.values[d] == 650.0 * _M
        assert tl.sources[d].startswith("imputed-plug: total_liabilities_and_equity - ")
        assert d not in _by_tag(rows, "temporary_equity").values

    def test_liabilities_plug_without_mezzanine(self):
        d = "2023-12-31"
        rows = self._le_rows(d, None, None)
        _apply_hierarchical_articulation(rows, {d})
        tl = _by_tag(rows, "total_liabilities")
        assert tl.values[d] == 650.0 * _M
        assert tl.sources[d].startswith("imputed-plug: total_liabilities_and_equity - ")
        assert d not in _by_tag(rows, "temporary_equity").values

    def test_mezzanine_remainder_with_reported_liabilities(self):
        d = "2023-12-31"
        rows = self._le_rows(d, 600.0, None)
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "total_liabilities").values[d] == 600.0 * _M
        assert _by_tag(rows, "temporary_equity").values[d] == 50.0 * _M

    def test_complete_rollup_section_keeps_its_amount(self):
        d = "2023-12-31"

        def inst(tag, value, parent=None, seq=1):
            return _rr(
                tag,
                {} if value is None else {d: value * _M},
                parent=parent,
                period_type="instant",
                balance="debit",
                sequence=seq,
                sources={} if value is None else {d: f"us-gaap:{tag}"},
            )

        rows = [
            inst("total_assets", 1000.0, seq=10),
            inst("total_current_assets", None, "total_assets", 1),
            inst("cash_and_equivalents", 300.0, "total_current_assets", 1),
            inst("accounts_receivable", 200.0, "total_current_assets", 2),
            inst("other_current_assets", 100.0, "total_current_assets", 3),
            inst("total_noncurrent_assets", 350.0, "total_assets", 5),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "total_current_assets").values[d] == 600.0 * _M
        noncurrent = _by_tag(rows, "total_noncurrent_assets")
        assert noncurrent.values[d] == 400.0 * _M
        assert noncurrent.sources[d].startswith("imputed-plug: total_assets - ")

    def test_repeated_amounts_contained_together(self):
        d = "2023-12-31"

        def inst(tag, value, seq, source):
            return _rr(
                tag,
                {d: value * _M},
                parent="total_liabilities",
                period_type="instant",
                balance="credit",
                sequence=seq,
                sources={d: source},
            )

        rows = [
            _rr(
                "total_liabilities",
                {d: 520.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=10,
                sources={d: "us-gaap:Liabilities"},
            ),
            inst("claims_and_claim_expenses", 300.0, 1, "us-gaap:A"),
            inst("future_policy_benefits", 300.0, 2, "us-gaap:B"),
            inst("policyholder_funds", 100.0, 3, "us-gaap:C"),
            inst("participating_policyholder_equity", 100.0, 4, "us-gaap:C"),
        ]
        _apply_hierarchical_articulation(rows, {d})
        assert _by_tag(rows, "future_policy_benefits").factor_on(d) == "0"
        assert _by_tag(rows, "participating_policyholder_equity").factor_on(d) == "0"
        assert _by_tag(rows, "claims_and_claim_expenses").factor_on(d) == "+"
        other = _by_tag(rows, "other_liabilities")
        assert other.values[d] == 120.0 * _M


class TestResolveSignFlips:
    @staticmethod
    def _equity(treasury, source="us-gaap:TreasuryStockValue", parent=100):
        d = _D
        return [
            _rr(
                "total_common_equity",
                {} if parent is None else {d: parent * _M},
                period_type="instant",
                sequence=10,
                sources={} if parent is None else {d: "us-gaap:StockholdersEquity"},
            ),
            _rr(
                "additional_paid_in_capital",
                {d: 140 * _M},
                parent="total_common_equity",
                period_type="instant",
                sources={d: "us-gaap:AdditionalPaidInCapital"},
            ),
            _rr(
                "treasury_stock",
                {d: treasury * _M},
                parent="total_common_equity",
                factor="-",
                period_type="instant",
                sequence=2,
                sources={d: source},
            ),
        ]

    @staticmethod
    def _facts(*values):
        entries = [
            _inst(_D, v * _M, filed=f"202{4 + i}-02-15") for i, v in enumerate(values)
        ]
        entries += [{"end": _D, "val": None}, {"val": 5}]
        return {"us-gaap": {"TreasuryStockValue": {"units": {"USD": entries}}}}

    def test_opposite_signed_fact_resolves_parent(self):
        rows = self._equity(-40)
        _resolve_sign_flips(rows, {_D}, self._facts(-40, 40), "annual", set(), [])
        treasury = _by_tag(rows, "treasury_stock")
        assert treasury.values[_D] == 40 * _M
        assert treasury.sources[_D] == "us-gaap:TreasuryStockValue(sign-resolved)"

    def test_parent_derived_by_rule(self):
        rows = self._equity(-40, parent=None) + [
            _rr("total_equity", {_D: 110 * _M}, period_type="instant"),
            _rr("total_preferred_equity", {_D: 10 * _M}, period_type="instant"),
        ]
        rules = [
            ("total_common_equity", [("noncontrolling_interests", 1)]),
            (
                "total_common_equity",
                [("total_equity", 1), ("total_preferred_equity", -1)],
            ),
        ]
        _resolve_sign_flips(rows, {_D}, self._facts(-40, 40), "annual", set(), rules)
        assert _by_tag(rows, "treasury_stock").values[_D] == 40 * _M

    @pytest.mark.parametrize(
        ("treasury", "source", "facts", "parent"),
        [
            (-40, "us-gaap:TreasuryStockValue", (-40,), 100),
            (-40, "us-gaap:TreasuryStockValue", (40,), 100),
            (-40, "imputed: total_equity - x", (-40, 40), 100),
            (0, "us-gaap:TreasuryStockValue", (0,), 140),
            (40, "us-gaap:TreasuryStockValue", (-40, 40), 100),
            (-40, "us-gaap:TreasuryStockValue", (-40, 40), None),
        ],
    )
    def test_sign_kept(self, treasury, source, facts, parent):
        rows = self._equity(treasury, source, parent)
        _resolve_sign_flips(rows, {_D}, self._facts(*facts), "annual", set(), [])
        assert _by_tag(rows, "treasury_stock").values[_D] == treasury * _M

    def test_two_candidate_children_left_unchanged(self):
        rows = self._equity(-40)
        rows.append(
            _rr(
                "retained_earnings",
                {_D: -40 * _M},
                parent="total_common_equity",
                factor="-",
                period_type="instant",
                sources={_D: "us-gaap:TreasuryStockValue"},
            )
        )
        rows[0].values[_D] = 140 * _M
        _resolve_sign_flips(rows, {_D}, self._facts(-40, 40), "annual", set(), [])
        assert _by_tag(rows, "treasury_stock").values[_D] == -40 * _M

    @staticmethod
    def _income(later):
        d, q = "2023-03-31", "2023-06-30"
        sources = {d: "us-gaap:IncomeLossFromDiscontinuedOperationsNetOfTax"}
        if later is not None:
            sources[later] = (
                "ytd_derived(us-gaap:IncomeLossFromDiscontinuedOperationsNetOfTax)"
            )
        rows = [
            _rr(
                "net_income",
                {d: 107 * _M},
                sequence=10,
                sources={d: "us-gaap:ProfitLoss"},
            ),
            _rr(
                "net_income_continuing",
                {d: 415 * _M},
                parent="net_income",
                sources={d: "us-gaap:IncomeLossFromContinuingOperations"},
            ),
            _rr(
                "net_income_discontinued",
                {d: 308 * _M, q: 1 * _M},
                parent="net_income",
                sequence=2,
                sources=sources,
            ),
        ]
        facts = {
            "us-gaap": {
                "IncomeLossFromDiscontinuedOperationsNetOfTax": {
                    "units": {
                        "USD": [
                            _dur(
                                d,
                                "2023-01-01",
                                -308 * _M,
                                form="10-Q",
                                filed="2023-05-01",
                            ),
                            _dur(
                                d,
                                "2023-01-01",
                                308 * _M,
                                form="10-Q",
                                filed="2024-05-01",
                            ),
                            _dur(
                                d,
                                "2022-04-01",
                                308 * _M,
                                form="10-K",
                                filed="2024-02-01",
                            ),
                        ]
                    }
                }
            }
        }
        return rows, facts, d

    @pytest.mark.parametrize(
        ("later", "expected"),
        [(None, -308), ("2023-06-30", 308), ("2024-06-30", -308)],
    )
    def test_quarterly_year_to_date_guard(self, later, expected):
        rows, facts, d = self._income(later)
        _resolve_sign_flips(rows, {d}, facts, "quarterly", set(), [])
        assert _by_tag(rows, "net_income_discontinued").values[d] == expected * _M

    def test_cash_flow_negated_row(self):
        d = _D
        rows = [
            _rr("net_cash_from_financing_activities", {d: -100 * _M}, sequence=10),
            _rr(
                "payment_of_dividends",
                {d: -50 * _M},
                parent="net_cash_from_financing_activities",
                sources={d: "us-gaap:PaymentsOfDividends"},
            ),
            _rr(
                "repurchase_of_common_equity",
                {d: 50 * _M},
                parent="net_cash_from_financing_activities",
                sequence=2,
                sources={d: "us-gaap:PaymentsForRepurchaseOfCommonStock"},
            ),
        ]
        facts = {
            "us-gaap": {
                "PaymentsForRepurchaseOfCommonStock": {
                    "units": {
                        "USD": [
                            _dur(d, "2023-01-01", -50 * _M),
                            _dur(d, "2023-01-01", 50 * _M, filed="2025-02-15"),
                        ]
                    }
                }
            }
        }
        _resolve_sign_flips(
            rows,
            {d},
            facts,
            "annual",
            {"payment_of_dividends", "repurchase_of_common_equity"},
            [],
        )
        repurchase = _by_tag(rows, "repurchase_of_common_equity")
        assert repurchase.values[d] == -50 * _M
        assert repurchase.sources[d].endswith("(sign-resolved)")


# ---------------------------------------------------------------------------
# impute() -- empty-ruleset early return
# ---------------------------------------------------------------------------


_QUARTERS = ["2023-03-31", "2023-06-30", "2023-09-30"]


def _concept_facts(fy, quarters, nine=()):
    starts = ["2023-01-01", "2023-04-01", "2023-07-01"]
    entries = [_dur(_D, "2023-01-01", v * _M, filed=f) for f, v in fy]
    entries += [
        _dur(_QUARTERS[2], "2023-01-01", v * _M, form="10-Q", filed=f) for f, v in nine
    ]
    entries += [
        _dur(d, s, v * _M, form="10-Q", filed="2024-11-01")
        for d, s, v in zip(_QUARTERS, starts, quarters)
    ]
    entries.append({"end": _D, "val": None})
    return {"units": {"USD": entries}}


class TestReconcileFiscalYearEnds:
    def test_annual_instant_row_created_in_quarterly(self):
        annual = StatementResult(
            statement="balance_sheet",
            company_type="industrial",
            frequency="annual",
            currency="USD",
            dates=[_D],
            rows=[
                _rr(
                    "goodwill",
                    {_D: 50.0},
                    period_type="instant",
                    sources={_D: "us-gaap:Goodwill"},
                )
            ],
        )
        quarterly = StatementResult(
            statement="balance_sheet",
            company_type="industrial",
            frequency="quarterly",
            currency="USD",
            dates=["2023-09-30", _D],
            rows=[],
        )
        reconcile_fiscal_year_ends(quarterly, annual, {"goodwill"})
        row = _by_tag(quarterly.rows, "goodwill")
        assert row.values == {_D: 50.0}
        assert row.sources == {_D: "us-gaap:Goodwill"}

    def test_fourth_quarter_only_remainder_row_holds_no_value(self):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]
        annual = StatementResult(
            statement="income_statement",
            company_type="industrial",
            frequency="annual",
            currency="USD",
            dates=[_D],
            rows=[
                _rr(
                    "total_x", {_D: 1000.0 * _M}, sequence=10, sources={_D: "us-gaap:X"}
                ),
                _rr(
                    "a",
                    {_D: 700.0 * _M},
                    parent="total_x",
                    sequence=1,
                    sources={_D: "us-gaap:A"},
                ),
            ],
        )
        quarterly = StatementResult(
            statement="income_statement",
            company_type="industrial",
            frequency="quarterly",
            currency="USD",
            dates=[*quarters, _D],
            rows=[
                _rr(
                    "total_x",
                    {q: 200.0 * _M for q in quarters},
                    sequence=10,
                    sources={q: "us-gaap:X" for q in quarters},
                ),
                _rr(
                    "a",
                    {q: 150.0 * _M for q in quarters},
                    parent="total_x",
                    sequence=1,
                    sources={q: "us-gaap:A" for q in quarters},
                ),
            ],
        )
        reconcile_fiscal_year_ends(quarterly, annual, {"total_x", "a"})
        assert _by_tag(quarterly.rows, "total_x").values[_D] == 400.0 * _M
        assert _by_tag(quarterly.rows, "a").values[_D] == 250.0 * _M
        other = _by_tag(quarterly.rows, "other_x")
        assert _D not in other.values

    def test_preliminary_dates_prefixed_once(self):
        d = "2024-03-31"
        annual = StatementResult(
            statement="income_statement",
            company_type="industrial",
            frequency="annual",
            currency="USD",
            dates=[_D],
            rows=[],
        )
        quarterly = StatementResult(
            statement="income_statement",
            company_type="industrial",
            frequency="quarterly",
            currency="USD",
            dates=[d],
            rows=[
                _rr("total_revenue", {d: 10.0}, sources={d: "us-gaap:Revenues"}),
                _rr("other_revenue", {d: 1.0}, sources={d: "preliminary:us-gaap:X"}),
                _rr("total_cost_of_revenue", {}),
            ],
            preliminary_dates={d},
        )
        reconcile_fiscal_year_ends(quarterly, annual, set())
        assert _by_tag(quarterly.rows, "total_revenue").sources[d] == (
            "preliminary:us-gaap:Revenues"
        )
        assert _by_tag(quarterly.rows, "other_revenue").sources[d] == (
            "preliminary:us-gaap:X"
        )
        assert _by_tag(quarterly.rows, "total_cost_of_revenue").sources == {}

    @staticmethod
    def _pair(statement, annual_rows, quarterly_rows, quarters):
        annual = StatementResult(
            statement=statement,
            company_type="industrial",
            frequency="annual",
            currency="USD",
            dates=[_D],
            rows=annual_rows,
        )
        quarterly = StatementResult(
            statement=statement,
            company_type="industrial",
            frequency="quarterly",
            currency="USD",
            dates=[*quarters, _D],
            rows=quarterly_rows,
        )
        return annual, quarterly

    _Q = _QUARTERS

    def _vintage(self, revenue_q, revenue_fy, facts, statement="income_statement"):
        tag = "us-gaap:Revenues"
        cost = "us-gaap:CostOfRevenue"
        gp = "imputed: total_revenue - total_cost_of_revenue"
        annual, quarterly = self._pair(
            statement,
            [
                _rr("total_revenue", {_D: revenue_fy * _M}, sources={_D: tag}),
                _rr("total_cost_of_revenue", {_D: 300 * _M}, sources={_D: cost}),
                _rr(
                    "total_gross_profit",
                    {_D: (revenue_fy - 300) * _M},
                    sources={_D: gp},
                ),
            ],
            [
                _rr(
                    "total_revenue",
                    {d: v * _M for d, v in zip(self._Q, revenue_q)},
                    sources=dict.fromkeys(self._Q, tag),
                ),
                _rr(
                    "total_cost_of_revenue",
                    dict.fromkeys(self._Q, 100 * _M),
                    sources=dict.fromkeys(self._Q, cost),
                ),
                _rr(
                    "total_gross_profit",
                    {d: (v - 100) * _M for d, v in zip(self._Q, revenue_q)},
                    sources=dict.fromkeys(self._Q, gp),
                ),
            ],
            self._Q,
        )
        tags = {"total_revenue", "total_cost_of_revenue", "total_gross_profit"}
        reconcile_fiscal_year_ends(quarterly, annual, tags, facts)
        return {tag: _by_tag(quarterly.rows, tag) for tag in tags}

    def test_q4_from_annual_filing_before_restatement(self):
        facts = {
            "us-gaap": {
                "Revenues": _concept_facts(
                    [("2025-02-15", 531), ("2024-02-15", 800)],
                    [200, 200, 200],
                    [("2023-11-01", 600)],
                ),
                "CostOfRevenue": _concept_facts(
                    [("2025-02-15", 300), ("2024-02-15", 400)],
                    [100, 100, 100],
                    [("2023-11-01", 300)],
                ),
            }
        }
        rows = self._vintage([200, 200, 200], 531, facts)
        assert rows["total_revenue"].values[_D] == 200 * _M
        assert "(filed 2024-02-15)" in rows["total_revenue"].sources[_D]
        assert rows["total_cost_of_revenue"].values[_D] == 100 * _M
        assert rows["total_gross_profit"].values[_D] == 100 * _M
        assert rows["total_gross_profit"].sources[_D] == (
            "imputed: total_revenue - total_cost_of_revenue"
        )

    def test_q4_from_latest_nine_months(self):
        facts = {
            "us-gaap": {
                "Revenues": _concept_facts(
                    [("2024-02-15", 600)], [300, 200, 200], [("2023-11-01", 500)]
                )
            }
        }
        rows = self._vintage([300, 200, 200], 600, facts)
        assert rows["total_revenue"].values[_D] == 100 * _M
        assert rows["total_revenue"].sources[_D] == (
            "Q4: FY[us-gaap:Revenues] \u2212 9M[us-gaap:Revenues]"
        )

    def test_q4_kept_without_opposite_signed_quarter(self):
        facts = {
            "us-gaap": {
                "Revenues": _concept_facts(
                    [("2025-02-15", 531), ("2024-02-15", 800)],
                    [150, 150, 150],
                    [("2023-11-01", 450)],
                )
            }
        }
        rows = self._vintage([150, 150, 150], 531, facts)
        assert rows["total_revenue"].values[_D] == 81 * _M

    @pytest.mark.parametrize(
        "facts",
        [
            None,
            {},
            {"us-gaap": {"Revenues": {"units": {"USD": []}}}},
            {
                "us-gaap": {
                    "Revenues": {
                        "units": {
                            "USD": [
                                _dur(_D, "2023-01-01", 531 * _M, filed="2025-02-15")
                            ]
                        }
                    }
                }
            },
            {
                "us-gaap": {
                    "Revenues": {
                        "units": {
                            "USD": [
                                _dur(_D, "2023-01-01", 531 * _M, filed="2025-02-15"),
                                _dur(_D, "2023-01-01", 800 * _M, filed="2024-02-15"),
                                _dur(
                                    "2023-09-30",
                                    "2023-01-01",
                                    600 * _M,
                                    form="10-Q",
                                    filed="2024-11-01",
                                ),
                            ]
                        }
                    }
                }
            },
            {
                "us-gaap": {
                    "Revenues": _concept_facts(
                        [("2025-02-15", 531), ("2024-02-15", 800)],
                        [200, 200, 200],
                        [("2023-11-01", 590)],
                    )
                }
            },
        ],
    )
    def test_q4_kept_without_filing_evidence(self, facts):
        rows = self._vintage([200, 200, 200], 531, facts)
        assert rows["total_revenue"].values[_D] == -69 * _M

    def test_q4_from_facts_of_negated_cash_flow_row(self):
        tag = "us-gaap:PaymentsToAcquirePropertyPlantAndEquipment"
        annual, quarterly = self._pair(
            "cash_flow",
            [
                _rr(
                    "purchase_of_plant_property_and_equipment",
                    {_D: -531 * _M},
                    balance="credit",
                    sources={_D: tag},
                )
            ],
            [
                _rr(
                    "purchase_of_plant_property_and_equipment",
                    dict.fromkeys(self._Q, -200 * _M),
                    balance="credit",
                    sources=dict.fromkeys(self._Q, tag),
                )
            ],
            self._Q,
        )
        facts = {
            "us-gaap": {
                "PaymentsToAcquirePropertyPlantAndEquipment": _concept_facts(
                    [("2025-02-15", 531), ("2024-02-15", 800)],
                    [200, 200, 200],
                    [("2023-11-01", 600)],
                )
            }
        }
        reconcile_fiscal_year_ends(
            quarterly, annual, {"purchase_of_plant_property_and_equipment"}, facts
        )
        row = _by_tag(quarterly.rows, "purchase_of_plant_property_and_equipment")
        assert row.values[_D] == -200 * _M

    def test_formula_terms(self):
        assert _formula_terms("imputed: -a + b - c") == [("a", -1), ("b", 1), ("c", -1)]

    def test_q4_from_nine_months_kept(self):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]
        tag = "us-gaap:PaymentsToAcquireBusinessesNetOfCashAcquired"
        nine = f"Q4: FY[{tag}] \u2212 9M[{tag}]"
        annual, quarterly = self._pair(
            "cash_flow",
            [_rr("acquisitions", {_D: -90.0 * _M}, sources={_D: tag})],
            [
                _rr(
                    "acquisitions",
                    {quarters[2]: -50.0 * _M, _D: -20.0 * _M},
                    sources={quarters[2]: f"ytd_derived({tag})", _D: nine},
                )
            ],
            quarters,
        )
        reconcile_fiscal_year_ends(quarterly, annual, {"acquisitions"})
        row = _by_tag(quarterly.rows, "acquisitions")
        assert row.values[_D] == -20.0 * _M
        assert row.sources[_D] == nine

    def test_q4_not_derived_from_partial_quarters(self):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]
        annual, quarterly = self._pair(
            "income_statement",
            [_rr("restructuring_charge", {_D: 90.0 * _M}, sources={_D: "us-gaap:R"})],
            [
                _rr(
                    "restructuring_charge",
                    {quarters[1]: 20.0 * _M, quarters[2]: 50.0 * _M},
                    sources={quarters[1]: "us-gaap:R", quarters[2]: "us-gaap:R"},
                )
            ],
            quarters,
        )
        reconcile_fiscal_year_ends(quarterly, annual, {"restructuring_charge"})
        assert _D not in _by_tag(quarterly.rows, "restructuring_charge").values

    def test_h2_value_cleared_in_quarterly_year(self):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]
        annual, quarterly = self._pair(
            "income_statement",
            [_rr("impairment_expense", {_D: 0.0}, sources={_D: "us-gaap:A"})],
            [
                _rr(
                    "impairment_expense",
                    {quarters[1]: 33.0 * _M, _D: -33.0 * _M},
                    sources={
                        quarters[1]: "us-gaap:A",
                        _D: "H2: FY[us-gaap:A] \u2212 H1[us-gaap:A]",
                    },
                )
            ],
            quarters,
        )
        reconcile_fiscal_year_ends(quarterly, annual, {"impairment_expense"})
        row = _by_tag(quarterly.rows, "impairment_expense")
        assert _D not in row.values
        assert _D not in row.sources

    def test_period_factor_copied_from_annual(self):
        annual_pref = _rr(
            "total_preferred_equity",
            {_D: 100.0 * _M},
            period_type="instant",
            sources={_D: "us-gaap:PreferredStockValue"},
        )
        annual_pref.date_factors[_D] = "0"
        annual_gw = _rr(
            "goodwill",
            {_D: 50.0 * _M},
            period_type="instant",
            sources={_D: "us-gaap:Goodwill"},
        )
        quarterly_gw = _rr(
            "goodwill", {_D: 40.0 * _M}, period_type="instant", sources={_D: "x"}
        )
        quarterly_gw.date_factors[_D] = "0"
        quarterly_dropped = _rr(
            "intangible_assets", {_D: 5.0 * _M}, period_type="instant"
        )
        quarterly_dropped.date_factors[_D] = "0"
        annual, quarterly = self._pair(
            "balance_sheet",
            [annual_pref, annual_gw],
            [
                _rr(
                    "total_preferred_equity",
                    {_D: 90.0 * _M},
                    period_type="instant",
                ),
                quarterly_gw,
                quarterly_dropped,
            ],
            ["2023-09-30"],
        )
        reconcile_fiscal_year_ends(quarterly, annual, set())
        assert _by_tag(quarterly.rows, "total_preferred_equity").factor_on(_D) == "0"
        gw = _by_tag(quarterly.rows, "goodwill")
        assert gw.values[_D] == 50.0 * _M
        assert _D not in gw.date_factors
        assert _D not in _by_tag(quarterly.rows, "intangible_assets").date_factors

    def test_q4_beginning_cash_follows_cash_identity(self):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]
        annual, quarterly = self._pair(
            "cash_flow",
            [
                _rr(
                    "net_change_in_cash",
                    {_D: 100.0 * _M},
                    sources={_D: "us-gaap:CashPeriodIncreaseDecrease"},
                ),
                _rr(
                    "cash_at_end_of_period",
                    {_D: 420.0 * _M},
                    period_type="instant",
                    sources={_D: "us-gaap:Cash"},
                ),
            ],
            [
                _rr(
                    "net_change_in_cash",
                    {q: v * _M for q, v in zip(quarters, (10.0, 20.0, 30.0))},
                    sources={q: "us-gaap:CashPeriodIncreaseDecrease" for q in quarters},
                ),
                _rr(
                    "cash_at_end_of_period",
                    {quarters[2]: 300.0 * _M},
                    period_type="instant",
                    sources={quarters[2]: "us-gaap:Cash"},
                ),
                _rr(
                    "cash_at_beginning_of_period",
                    {_D: 300.0 * _M},
                    period_type="instant",
                    sources={_D: f"derived: cash_at_end_of_period({quarters[2]})"},
                ),
            ],
            quarters,
        )
        reconcile_fiscal_year_ends(quarterly, annual, {"net_change_in_cash"})
        assert _by_tag(quarterly.rows, "net_change_in_cash").values[_D] == 40.0 * _M
        bop = _by_tag(quarterly.rows, "cash_at_beginning_of_period")
        assert bop.values[_D] == 380.0 * _M
        assert bop.sources[_D] == (
            "identity-enforced: cash_at_end_of_period - net_change_in_cash"
        )

    def test_q4_rollup_quarters_follow_q4_components(self):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]
        annual, quarterly = self._pair(
            "cash_flow",
            [
                _rr(
                    "depreciation_and_amortization",
                    {_D: 100.0 * _M},
                    sources={_D: "us-gaap:DepreciationDepletionAndAmortization"},
                ),
                _rr(
                    "depreciation_expense",
                    {_D: 60.0 * _M},
                    parent="depreciation_and_amortization",
                    sources={_D: "us-gaap:Depreciation"},
                ),
                _rr(
                    "amortization_expense",
                    {_D: 40.0 * _M},
                    parent="depreciation_and_amortization",
                    sources={_D: "us-gaap:AmortizationOfIntangibleAssets"},
                ),
            ],
            [
                _rr(
                    "depreciation_and_amortization",
                    {q: 12.0 * _M for q in quarters},
                    sources={
                        q: "imputed-rollup: depreciation_expense(+)" for q in quarters
                    },
                ),
                _rr(
                    "depreciation_expense",
                    {q: 12.0 * _M for q in quarters},
                    parent="depreciation_and_amortization",
                    sources={q: "us-gaap:Depreciation" for q in quarters},
                ),
            ],
            quarters,
        )
        reconcile_fiscal_year_ends(
            quarterly,
            annual,
            {"depreciation_and_amortization", "depreciation_expense"},
        )
        da = _by_tag(quarterly.rows, "depreciation_and_amortization")
        assert _by_tag(quarterly.rows, "depreciation_expense").values[_D] == 24.0 * _M
        assert da.values[_D] == 24.0 * _M
        assert da.sources[_D].startswith("imputed-rollup")

    @pytest.mark.parametrize(
        ("fy_value", "fy_source", "amortization", "expected"),
        [
            (150.0, "us-gaap:DepreciationDepletionAndAmortization", 40.0, 40.0),
            (60.0, "imputed-rollup: depreciation_expense(+)", None, 24.0),
        ],
    )
    def test_q4_of_rolled_up_quarters_rolls_up(
        self, fy_value, fy_source, amortization, expected
    ):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]
        annual_rows = [
            _rr(
                "depreciation_and_amortization",
                {_D: fy_value * _M},
                sources={_D: fy_source},
            ),
            _rr(
                "depreciation_expense",
                {_D: 60.0 * _M},
                parent="depreciation_and_amortization",
                sources={_D: "us-gaap:Depreciation"},
            ),
        ]
        if amortization is not None:
            annual_rows.append(
                _rr(
                    "amortization_expense",
                    {_D: amortization * _M},
                    parent="depreciation_and_amortization",
                    sources={_D: "us-gaap:AmortizationOfIntangibleAssets"},
                )
            )
        annual, quarterly = self._pair(
            "cash_flow",
            annual_rows,
            [
                _rr(
                    "depreciation_and_amortization",
                    {q: 20.0 * _M for q in quarters},
                    sources={
                        q: "imputed-rollup: depreciation_expense(+)"
                        " + amortization_expense(+)"
                        for q in quarters
                    },
                ),
                _rr(
                    "depreciation_expense",
                    {q: 12.0 * _M for q in quarters},
                    parent="depreciation_and_amortization",
                    sources={q: "us-gaap:Depreciation" for q in quarters},
                ),
                _rr(
                    "amortization_expense",
                    {q: 8.0 * _M for q in quarters},
                    parent="depreciation_and_amortization",
                    sources={
                        q: "us-gaap:AmortizationOfIntangibleAssets" for q in quarters
                    },
                ),
            ],
            quarters,
        )
        reconcile_fiscal_year_ends(
            quarterly,
            annual,
            {
                "depreciation_and_amortization",
                "depreciation_expense",
                "amortization_expense",
            },
        )
        da = _by_tag(quarterly.rows, "depreciation_and_amortization")
        assert da.values[_D] == expected * _M
        assert da.sources[_D].startswith("imputed-rollup")

    def test_q4_of_rolled_up_quarters_with_annual_remainder(self):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]
        annual, quarterly = self._pair(
            "cash_flow",
            [
                _rr(
                    "depreciation_and_amortization",
                    {_D: 150.0 * _M},
                    sources={_D: "us-gaap:DepreciationDepletionAndAmortization"},
                ),
                _rr(
                    "depreciation_expense",
                    {_D: 60.0 * _M},
                    parent="depreciation_and_amortization",
                    sources={_D: "us-gaap:Depreciation"},
                ),
                _rr(
                    "other_depreciation_and_amortization",
                    {_D: 90.0 * _M},
                    parent="depreciation_and_amortization",
                    sources={
                        _D: "imputed-plug: depreciation_and_amortization"
                        " - (depreciation_expense(+))"
                    },
                ),
            ],
            [
                _rr(
                    "depreciation_and_amortization",
                    {q: 30.0 * _M for q in quarters},
                    sources={
                        q: "imputed-rollup: depreciation_expense(+)"
                        " + other_depreciation_and_amortization(+)"
                        for q in quarters
                    },
                ),
                _rr(
                    "depreciation_expense",
                    {q: 12.0 * _M for q in quarters},
                    parent="depreciation_and_amortization",
                    sources={q: "us-gaap:Depreciation" for q in quarters},
                ),
                _rr(
                    "other_depreciation_and_amortization",
                    {q: 18.0 * _M for q in quarters},
                    parent="depreciation_and_amortization",
                    sources={
                        q: "us-gaap:OtherDepreciationAndAmortization" for q in quarters
                    },
                ),
            ],
            quarters,
        )
        reconcile_fiscal_year_ends(
            quarterly,
            annual,
            {
                "depreciation_and_amortization",
                "depreciation_expense",
                "other_depreciation_and_amortization",
            },
        )
        da = _by_tag(quarterly.rows, "depreciation_and_amortization")
        assert da.values[_D] == 60.0 * _M
        assert da.sources[_D].startswith("Q4: FY[")

    def test_q4_from_quarters_concept_kept(self):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]
        q4 = "Q4: FY[us-gaap:Revenues] \u2212 (Q1[us-gaap:Revenues])"
        annual, quarterly = self._pair(
            "income_statement",
            [_rr("operating_revenue", {_D: 700.0 * _M}, sources={_D: "us-gaap:R"})],
            [
                _rr(
                    "operating_revenue",
                    {**{q: 200.0 * _M for q in quarters}, _D: 400.0 * _M},
                    sources={**{q: "us-gaap:Revenues" for q in quarters}, _D: q4},
                ),
                _rr(
                    "total_revenue",
                    {q: 200.0 * _M for q in quarters},
                    sources={q: "us-gaap:Revenues" for q in quarters},
                ),
            ],
            quarters,
        )
        annual.rows.append(
            _rr("total_revenue", {_D: 700.0 * _M}, sources={_D: "us-gaap:R"})
        )
        reconcile_fiscal_year_ends(
            quarterly, annual, {"operating_revenue", "total_revenue"}
        )
        row = _by_tag(quarterly.rows, "operating_revenue")
        assert row.values[_D] == 400.0 * _M
        assert row.sources[_D] == q4
        assert _by_tag(quarterly.rows, "total_revenue").values[_D] == 100.0 * _M

    @pytest.mark.parametrize("without_q4", [False, True])
    def test_zero_q4_remainder_written(self, without_q4):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]
        plug = "imputed-plug: total_x - (a(+))"
        annual, quarterly = self._pair(
            "income_statement",
            [
                _rr(
                    "total_x", {_D: 1000.0 * _M}, sequence=10, sources={_D: "us-gaap:X"}
                ),
                _rr(
                    "a",
                    {_D: 700.0 * _M},
                    parent="total_x",
                    sequence=1,
                    sources={_D: "us-gaap:A"},
                ),
                _rr(
                    "other_x",
                    {_D: 300.0 * _M},
                    parent="total_x",
                    sequence=2,
                    sources={_D: plug},
                ),
            ],
            [
                _rr(
                    "total_x",
                    {q: 250.0 * _M for q in quarters},
                    sequence=10,
                    sources={q: "us-gaap:X" for q in quarters},
                ),
                _rr(
                    "a",
                    {q: 150.0 * _M for q in quarters},
                    parent="total_x",
                    sequence=1,
                    sources={q: "us-gaap:A" for q in quarters},
                ),
                _rr(
                    "other_x",
                    {**{q: 100.0 * _M for q in quarters}, _D: 5.0 * _M},
                    parent="total_x",
                    sequence=2,
                    sources={**{q: plug for q in quarters}, _D: plug},
                ),
            ],
            quarters,
        )
        if without_q4:
            other = _by_tag(quarterly.rows, "other_x")
            other.values.pop(_D)
            other.sources.pop(_D)
        reconcile_fiscal_year_ends(quarterly, annual, {"total_x", "a", "other_x"})
        other = _by_tag(quarterly.rows, "other_x")
        assert other.values[_D] == 0.0
        assert other.sources[_D].startswith("imputed-plug: total_x - ")

    @pytest.mark.parametrize(
        ("income", "fy_basic", "expected"),
        [
            (200, 101, (103_967_391, 105_967_391, 1.92, 1.89)),
            (-50, 101, (103_967_391, 103_967_391, -0.48, -0.48)),
            (200, 1, (None, 105_967_391, None, 1.89)),
        ],
    )
    def test_q4_per_share_derived(self, income, fy_basic, expected):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]

        def per_share(tag, fy, q, *, parent=None, factor="0", unit="per_share"):
            source = f"us-gaap:{tag}"
            return (
                _rr(
                    tag,
                    {_D: fy},
                    parent=parent,
                    factor=factor,
                    unit=unit,
                    sources={_D: source},
                ),
                _rr(
                    tag,
                    dict.fromkeys(quarters, q),
                    parent=parent,
                    factor=factor,
                    unit=unit,
                    sources=dict.fromkeys(quarters, source),
                ),
            )

        pairs = [
            per_share(
                "weighted_ave_basic_shares_os",
                fy_basic * _M,
                100 * _M,
                parent="basic_eps",
                factor="/",
                unit="shares",
            ),
            per_share(
                "weighted_ave_diluted_shares_os",
                103 * _M,
                102 * _M,
                parent="diluted_eps",
                factor="/",
                unit="shares",
            ),
            per_share("basic_eps", 2.0, 0.5),
            per_share("diluted_eps", 2.0, 0.5),
            per_share("cash_dividends_per_share", 0.4, 0.1),
        ]
        annual = StatementResult(
            statement="income_statement",
            company_type="industrial",
            frequency="annual",
            currency="USD",
            dates=["2022-12-31", _D],
            rows=[a for a, _ in pairs],
        )
        quarterly = StatementResult(
            statement="income_statement",
            company_type="industrial",
            frequency="quarterly",
            currency="USD",
            dates=[*quarters, _D],
            rows=[
                *(q for _, q in pairs),
                _rr(
                    "net_income_to_common",
                    {_D: income * _M},
                    sources={
                        _D: "us-gaap:NetIncomeLossAvailableToCommonStockholdersBasic"
                    },
                ),
            ],
        )
        reconcile_fiscal_year_ends(quarterly, annual, set())
        rows = {r.tag: r for r in quarterly.rows}
        assert (
            rows["weighted_ave_basic_shares_os"].values.get(_D),
            rows["weighted_ave_diluted_shares_os"].values.get(_D),
            rows["basic_eps"].values.get(_D),
            rows["diluted_eps"].values.get(_D),
        ) == expected
        assert rows["cash_dividends_per_share"].values[_D] == 0.1

    @pytest.mark.parametrize("interim", [True, False])
    def test_q4_balance_sheet_keeps_interim_lines(self, interim):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]
        annual, quarterly = self._pair(
            "balance_sheet",
            [
                _rr(
                    "total_assets",
                    {_D: 1000 * _M},
                    period_type="instant",
                    sources={_D: "us-gaap:Assets"},
                ),
                _rr(
                    "gross_ppe",
                    {_D: 500 * _M},
                    period_type="instant",
                    sources={_D: "us-gaap:PropertyPlantAndEquipmentGross"},
                ),
            ],
            [
                _rr(
                    "total_assets",
                    dict.fromkeys(quarters, 900 * _M) if interim else {},
                    period_type="instant",
                    sources=dict.fromkeys(quarters, "us-gaap:Assets")
                    if interim
                    else {},
                ),
            ],
            quarters,
        )
        reconcile_fiscal_year_ends(quarterly, annual, set())
        rows = {r.tag: r for r in quarterly.rows}
        assert rows["total_assets"].values[_D] == 1000 * _M
        assert (_D in rows["gross_ppe"].values) is not interim

    def test_q4_balance_sheet_remainder_needs_its_total(self):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]
        plug = "imputed-plug: total_current_liabilities - (accounts_payable(+))"
        annual, quarterly = self._pair(
            "balance_sheet",
            [
                _rr(
                    "total_current_liabilities",
                    {_D: 500 * _M},
                    period_type="instant",
                    sources={_D: "us-gaap:LiabilitiesCurrent"},
                ),
                _rr(
                    "other_current_liabilities",
                    {_D: 200 * _M},
                    parent="total_current_liabilities",
                    period_type="instant",
                    sources={_D: plug},
                ),
            ],
            [
                _rr(
                    "other_current_liabilities",
                    dict.fromkeys(quarters, 150 * _M),
                    parent="total_current_liabilities",
                    period_type="instant",
                    sources=dict.fromkeys(quarters, "us-gaap:OtherLiabilitiesCurrent"),
                ),
            ],
            quarters,
        )
        reconcile_fiscal_year_ends(quarterly, annual, set())
        rows = {r.tag: r for r in quarterly.rows}
        assert _D not in rows["total_current_liabilities"].values
        assert _D not in rows["other_current_liabilities"].values

    @pytest.mark.parametrize(
        ("source", "held"),
        [
            ("Q4: FY[us-gaap:OtherIncome] \u2212 (Q1[us-gaap:OtherIncome])", False),
            ("us-gaap:GoodwillAndIntangibleAssetImpairment", True),
        ],
    )
    def test_q4_only_line_held_when_reported(self, source, held):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]
        annual, quarterly = self._pair(
            "income_statement",
            [_rr("total_revenue", {_D: 400 * _M}, sources={_D: "us-gaap:Revenues"})],
            [
                _rr(
                    "total_revenue",
                    dict.fromkeys(quarters, 100 * _M),
                    sources=dict.fromkeys(quarters, "us-gaap:Revenues"),
                ),
                _rr("impairment_expense", {_D: 50 * _M}, sources={_D: source}),
            ],
            quarters,
        )
        reconcile_fiscal_year_ends(quarterly, annual, {"total_revenue"})
        rows = {r.tag: r for r in quarterly.rows}
        assert (_D in rows["impairment_expense"].values) is held

    def test_q4_instant_line_held_when_quarters_report_it(self):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]
        annual, quarterly = self._pair(
            "cash_flow",
            [
                _rr("net_change_in_cash", {_D: 40 * _M}, sources={_D: "us-gaap:N"}),
                _rr(
                    "cash_at_end_of_period",
                    {_D: 340 * _M},
                    period_type="instant",
                    sources={_D: "us-gaap:Cash"},
                ),
            ],
            [
                _rr(
                    "net_change_in_cash",
                    dict.fromkeys(quarters, 10 * _M),
                    sources=dict.fromkeys(quarters, "us-gaap:N"),
                ),
            ],
            quarters,
        )
        reconcile_fiscal_year_ends(quarterly, annual, {"net_change_in_cash"})
        rows = {r.tag: r for r in quarterly.rows}
        assert rows["net_change_in_cash"].values[_D] == 10 * _M
        assert _D not in rows["cash_at_end_of_period"].values

    def test_q4_only_remainder_not_held(self):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]
        plug = "imputed-plug: total_x - (a(+))"
        annual, quarterly = self._pair(
            "income_statement",
            [
                _rr(
                    "total_x", {_D: 1000.0 * _M}, sequence=10, sources={_D: "us-gaap:X"}
                ),
                _rr(
                    "a",
                    {_D: 700.0 * _M},
                    parent="total_x",
                    sequence=1,
                    sources={_D: "us-gaap:A"},
                ),
                _rr(
                    "other_x",
                    {_D: 300.0 * _M},
                    parent="total_x",
                    sequence=2,
                    sources={_D: plug},
                ),
            ],
            [
                _rr(
                    "total_x",
                    dict.fromkeys(quarters, 250.0 * _M),
                    sequence=10,
                    sources=dict.fromkeys(quarters, "us-gaap:X"),
                ),
                _rr(
                    "a",
                    dict.fromkeys(quarters, 250.0 * _M),
                    parent="total_x",
                    sequence=1,
                    sources=dict.fromkeys(quarters, "us-gaap:A"),
                ),
                _rr("other_x", {}, parent="total_x", sequence=2),
            ],
            quarters,
        )
        reconcile_fiscal_year_ends(quarterly, annual, {"total_x", "a", "other_x"})
        rows = {r.tag: r for r in quarterly.rows}
        assert rows["total_x"].values[_D] == 250.0 * _M
        assert rows["a"].values[_D] == -50.0 * _M
        assert _D not in rows["other_x"].values

    def test_q4_beginning_cash_kept_when_identity_holds(self):
        quarters = ["2023-03-31", "2023-06-30", "2023-09-30"]
        annual, quarterly = self._pair(
            "cash_flow",
            [
                _rr("net_change_in_cash", {_D: 100.0 * _M}, sources={_D: "us-gaap:N"}),
                _rr(
                    "cash_at_end_of_period",
                    {_D: 340.0 * _M},
                    period_type="instant",
                    sources={_D: "us-gaap:Cash"},
                ),
            ],
            [
                _rr(
                    "net_change_in_cash",
                    {q: v * _M for q, v in zip(quarters, (10.0, 20.0, 30.0))},
                    sources={q: "us-gaap:N" for q in quarters},
                ),
                _rr(
                    "cash_at_beginning_of_period",
                    {_D: 300.0 * _M},
                    period_type="instant",
                    sources={_D: "derived: cash_at_end_of_period(2023-09-30)"},
                ),
            ],
            quarters,
        )
        reconcile_fiscal_year_ends(quarterly, annual, {"net_change_in_cash"})
        bop = _by_tag(quarterly.rows, "cash_at_beginning_of_period")
        assert bop.values[_D] == 300.0 * _M
        assert bop.sources[_D].startswith("derived:")


class TestImputeEmptyRuleset:
    def test_empty_ruleset_returns_rows_unchanged(self):
        # When a statement's rule set is empty, impute() short-circuits and
        # returns the rows untouched with no diagnostics (line 247).
        rows = [_rr("total_assets", {_D: 100.0 * _M})]
        with patch("openbb_sec.utils.statement_schema._imputation.BS_IMPUTE", []):
            out, diags = impute(rows, "balance_sheet", "industrial", {_D})
        assert out is rows
        assert diags == []


# ---------------------------------------------------------------------------
# impute() -- income-statement pre-pass corrections (equity method / ProfitLoss)
# ---------------------------------------------------------------------------


class TestImputeEquityMethodPrePass:
    def test_identity_holds_skips_correction(self):
        # pretax tagged with EquityMethodInvestments but nic+tax already match ->
        # the correction block hits the early `continue` and leaves pretax as-is.
        rows = [
            _rr(
                "total_pretax_income",
                {_D: 550.0 * _M},
                sequence=1,
                sources={
                    _D: "us-gaap:IncomeLossFromContinuingOperationsBeforeIncomeTaxesMinorityInterestAndIncomeLossFromEquityMethodInvestments"
                },
            ),
            _rr(
                "income_tax_expense",
                {_D: 150.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:IncomeTaxExpenseBenefit"},
            ),
            _rr(
                "net_income_continuing",
                {_D: 400.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:IncomeLossFromContinuingOperations"},
            ),
        ]
        out, _ = impute(
            rows, "income_statement", "diversified", {_D}, facts={"us-gaap": {}}
        )
        ptx = _by_tag(out, "total_pretax_income")
        assert ptx.values[_D] == 550.0 * _M
        assert "IncomeLossFromContinuing" in ptx.sources[_D]

    def test_profitloss_nic_deletes_equity_method_pretax(self):
        # Identity off + nic from ProfitLoss (NCI-bearing, no FromContinuing) ->
        # pretax value/source are deleted (281-283).
        rows = [
            _rr(
                "total_pretax_income",
                {_D: 700.0 * _M},
                sequence=1,
                sources={
                    _D: "us-gaap:IncomeLossFromContinuingOperationsBeforeIncomeTaxesMinorityInterestAndIncomeLossFromEquityMethodInvestments"
                },
            ),
            _rr(
                "income_tax_expense",
                {_D: 150.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:IncomeTaxExpenseBenefit"},
            ),
            _rr(
                "net_income_continuing",
                {_D: 400.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:ProfitLoss"},
            ),
            _rr("income_before_equity_method", {}, sequence=4),
            _rr("equity_method_investments", {}, sequence=5),
        ]
        out, _ = impute(
            rows, "income_statement", "diversified", {_D}, facts={"us-gaap": {}}
        )
        ptx = _by_tag(out, "total_pretax_income")
        # pretax was deleted then may be re-imputed by IS_IMPUTE_COMMON from nic+tax.
        assert ptx.values.get(_D) in (None, 550.0 * _M)
        assert "EquityMethodInvestments" not in ptx.sources.get(_D, "")

    def test_no_equity_value_deletes_pretax(self):
        # Identity off, nic not ProfitLoss, equity_method_investments absent/zero ->
        # else-branch deletes pretax (294-295).
        rows = [
            _rr(
                "total_pretax_income",
                {_D: 900.0 * _M},
                sequence=1,
                sources={
                    _D: "us-gaap:IncomeLossFromContinuingOperationsBeforeIncomeTaxesExtraordinaryItemsNoncontrollingInterestEquityMethodInvestments"
                },
            ),
            _rr(
                "income_tax_expense",
                {_D: 150.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:IncomeTaxExpenseBenefit"},
            ),
            _rr(
                "net_income_continuing",
                {_D: 400.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:IncomeLossFromContinuingOperations"},
            ),
            _rr("income_before_equity_method", {}, sequence=4),
            _rr("equity_method_investments", {_D: 0.0}, sequence=5),
        ]
        out, _ = impute(
            rows, "income_statement", "diversified", {_D}, facts={"us-gaap": {}}
        )
        ptx = _by_tag(out, "total_pretax_income")
        # original EquityMethod-sourced value removed; re-imputed to nic+tax=550.
        assert "EquityMethodInvestments" not in ptx.sources.get(_D, "")

    def test_income_before_equity_seeded_from_pretax(self):
        # income_before_equity_method is empty -> seeded with pretax value/source (267-268).
        rows = [
            _rr(
                "total_pretax_income",
                {_D: 700.0 * _M},
                sequence=1,
                sources={
                    _D: "us-gaap:IncomeLossFromContinuingOperationsBeforeIncomeTaxesMinorityInterestAndIncomeLossFromEquityMethodInvestments"
                },
            ),
            _rr("income_before_equity_method", {}, sequence=2),
            _rr(
                "equity_method_investments",
                {_D: 50.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:IncomeLossFromEquityMethodInvestments"},
            ),
            _rr(
                "income_tax_expense",
                {_D: 150.0 * _M},
                sequence=4,
                sources={_D: "us-gaap:IncomeTaxExpenseBenefit"},
            ),
            _rr(
                "net_income_continuing",
                {_D: 600.0 * _M},
                sequence=5,
                sources={_D: "us-gaap:IncomeLossFromContinuingOperations"},
            ),
        ]
        out, _ = impute(
            rows, "income_statement", "diversified", {_D}, facts={"us-gaap": {}}
        )
        beq = _by_tag(out, "income_before_equity_method")
        assert beq.values[_D] == 700.0 * _M


class TestImputeProfitLossDiscAdjust:
    def test_profitloss_before_marker_skips(self):
        # nic source has ProfitLossBefore -> the disc-adjust block skips (312).
        rows = [
            _rr(
                "net_income_continuing",
                {_D: 400.0 * _M},
                sequence=1,
                sources={_D: "us-gaap:ProfitLossBeforeTax"},
            ),
            _rr("net_income_discontinued", {_D: 30.0 * _M}, sequence=2),
            _rr("income_tax_expense", {_D: 100.0 * _M}, sequence=3),
        ]
        out, _ = impute(
            rows, "income_statement", "industrial", {_D}, facts={"us-gaap": {}}
        )
        nic = _by_tag(out, "net_income_continuing")
        assert "(disc-adjusted)" not in nic.sources[_D]
        assert nic.values[_D] == 400.0 * _M

    def test_nonscoped_tax_skips_disc_adjust(self):
        # nic from ProfitLoss, but tax has a non-ContinuingOperations source ->
        # the block hits `continue` and disc is NOT subtracted (316-317).
        rows = [
            _rr(
                "net_income_continuing",
                {_D: 430.0 * _M},
                sequence=1,
                sources={_D: "us-gaap:ProfitLoss"},
            ),
            _rr("net_income_discontinued", {_D: 30.0 * _M}, sequence=2),
            _rr(
                "income_tax_expense",
                {_D: 100.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:IncomeTaxExpenseBenefit"},
            ),
        ]
        out, _ = impute(
            rows, "income_statement", "industrial", {_D}, facts={"us-gaap": {}}
        )
        nic = _by_tag(out, "net_income_continuing")
        assert "(disc-adjusted)" not in nic.sources[_D]
        assert nic.values[_D] == 430.0 * _M

    def test_disc_adjustment_applied_with_continuing_ops_tax(self):
        # nic from ProfitLoss (NCI/disc-bearing); tax explicitly ContinuingOperations
        # -> the guard passes and disc is subtracted (319-323).
        rows = [
            _rr(
                "net_income_continuing",
                {_D: 430.0 * _M},
                sequence=1,
                sources={_D: "us-gaap:ProfitLoss"},
            ),
            _rr("net_income_discontinued", {_D: 30.0 * _M}, sequence=2),
            _rr(
                "income_tax_expense",
                {_D: 100.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:IncomeTaxExpenseBenefitContinuingOperations"},
            ),
        ]
        out, _ = impute(
            rows, "income_statement", "industrial", {_D}, facts={"us-gaap": {}}
        )
        nic = _by_tag(out, "net_income_continuing")
        assert nic.values[_D] == 400.0 * _M  # 430 - 30
        assert "(disc-adjusted)" in nic.sources[_D]

    def test_disc_adjustment_applied_with_absent_tax_row(self):
        # nic from ProfitLoss; tax row absent -> no tax-source guard triggered,
        # disc subtraction still applies.
        rows = [
            _rr(
                "net_income_continuing",
                {_D: 430.0 * _M},
                sequence=1,
                sources={_D: "us-gaap:ProfitLoss"},
            ),
            _rr("net_income_discontinued", {_D: 30.0 * _M}, sequence=2),
        ]
        out, _ = impute(
            rows, "income_statement", "industrial", {_D}, facts={"us-gaap": {}}
        )
        nic = _by_tag(out, "net_income_continuing")
        assert nic.values[_D] == 400.0 * _M
        assert "(disc-adjusted)" in nic.sources[_D]


# ---------------------------------------------------------------------------
# impute() -- income-statement gross-profit / cogs / opex correction passes
# ---------------------------------------------------------------------------


class TestImputeBalanceSheetRound2:
    def test_gross_ppe_from_net_and_accumulated_depreciation(self):
        rows = [
            _rr(
                "net_ppe",
                {_D: 24675.0 * _M},
                period_type="instant",
                sequence=3,
                sources={_D: "us-gaap:PropertyPlantAndEquipmentNet"},
            ),
            _rr("gross_ppe", {}, parent="net_ppe", period_type="instant", sequence=1),
            _rr(
                "accumulated_depreciation",
                {_D: 12560.0 * _M},
                parent="net_ppe",
                factor="-",
                period_type="instant",
                sequence=2,
                sources={
                    _D: "us-gaap:AccumulatedDepreciationDepletionAndAmortizationPropertyPlantAndEquipment"
                },
            ),
        ]
        out, _ = impute(rows, "balance_sheet", "industrial", {_D}, facts={})
        gross = _by_tag(out, "gross_ppe")
        assert gross.values[_D] == 37235.0 * _M
        assert gross.sources[_D] == "imputed: net_ppe + accumulated_depreciation"
        assert _by_tag(out, "other_net_ppe") is None

    def test_equity_including_nci_corrected_from_balance_identity(self):
        def inst(tag, value, parent=None, seq=1):
            return _rr(
                tag,
                {} if value is None else {_D: value * _M},
                parent=parent,
                period_type="instant",
                sequence=seq,
                sources={} if value is None else {_D: f"us-gaap:{tag}"},
            )

        rows = [
            inst("total_liabilities_and_equity", 63519.4, seq=30),
            inst("total_liabilities", 52400.3, "total_liabilities_and_equity", 10),
            inst(
                "total_equity_and_noncontrolling_interests",
                -1808.5,
                "total_liabilities_and_equity",
                20,
            ),
            inst(
                "total_equity", 11119.1, "total_equity_and_noncontrolling_interests", 15
            ),
            inst(
                "noncontrolling_interests",
                None,
                "total_equity_and_noncontrolling_interests",
                16,
            ),
            inst("temporary_equity", None, "total_liabilities_and_equity", 21),
        ]
        out, _ = impute(rows, "balance_sheet", "insurance", {_D}, facts={})
        enci = _by_tag(out, "total_equity_and_noncontrolling_interests")
        assert enci.values[_D] == 11119.1 * _M
        assert enci.sources[_D] == (
            "corrected: total_equity + noncontrolling_interests"
        )
        assert _by_tag(out, "total_equity").values[_D] == 11119.1 * _M
        assert _D not in _by_tag(out, "temporary_equity").values

    def test_fallback_equity_including_nci_corrected_to_reference_equity(self):
        def inst(tag, value, parent=None, seq=1, source=None):
            return _rr(
                tag,
                {} if value is None else {_D: value * _M},
                parent=parent,
                period_type="instant",
                sequence=seq,
                sources={} if value is None else {_D: source or f"us-gaap:{tag}"},
            )

        rows = [
            inst("total_liabilities_and_equity", 9347.0, seq=30),
            inst(
                "total_liabilities",
                10874.0,
                "total_liabilities_and_equity",
                10,
                "imputed-plug: total_liabilities_and_equity - "
                "(total_equity_and_noncontrolling_interests(+))",
            ),
            inst(
                "total_equity_and_noncontrolling_interests",
                -1527.0,
                "total_liabilities_and_equity",
                20,
                "us-gaap:StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest(fallback)",
            ),
            inst(
                "total_equity",
                -1520.0,
                "total_equity_and_noncontrolling_interests",
                15,
                "us-gaap:StockholdersEquity",
            ),
            inst(
                "noncontrolling_interests",
                None,
                "total_equity_and_noncontrolling_interests",
                16,
            ),
            inst("temporary_equity", None, "total_liabilities_and_equity", 21),
        ]
        out, _ = impute(rows, "balance_sheet", "industrial", {_D}, facts={})
        enci = _by_tag(out, "total_equity_and_noncontrolling_interests")
        assert enci.values[_D] == -1520.0 * _M
        assert enci.sources[_D] == (
            "corrected: total_equity + noncontrolling_interests"
        )
        assert _by_tag(out, "total_equity").values[_D] == -1520.0 * _M
        assert _by_tag(out, "total_liabilities").values[_D] == 10867.0 * _M


class TestNetChangeFromBalances:
    def _rows(self, eop, eop_sources, ncc=None):
        return [
            _rr("net_change_in_cash", ncc or {}, sequence=1),
            _rr(
                "cash_at_end_of_period",
                eop,
                period_type="instant",
                sequence=2,
                sources=eop_sources,
            ),
        ]

    def test_annual_change_of_same_tag_balances(self):
        rows = self._rows(
            {"2022-12-31": 1000.0 * _M, "2023-12-31": 1200.0 * _M},
            {"2022-12-31": "us-gaap:Cash", "2023-12-31": "us-gaap:Cash(fallback)"},
        )
        _net_change_from_balances(rows, {"2022-12-31", "2023-12-31"}, "annual")
        ncc = _by_tag(rows, "net_change_in_cash")
        assert ncc.values == {"2023-12-31": 200.0 * _M}
        assert ncc.sources["2023-12-31"] == (
            "imputed: cash_at_end_of_period - cash_at_end_of_period(2022-12-31)"
        )

    def test_quarterly_span(self):
        rows = self._rows(
            {"2023-03-31": 1000.0 * _M, "2023-06-30": 900.0 * _M},
            {"2023-03-31": "us-gaap:Cash", "2023-06-30": "us-gaap:Cash"},
        )
        _net_change_from_balances(rows, {"2023-03-31", "2023-06-30"}, "quarterly")
        assert _by_tag(rows, "net_change_in_cash").values == {"2023-06-30": -100.0 * _M}

    def test_quarter_gap_not_annual_span(self):
        rows = self._rows(
            {"2023-03-31": 1000.0 * _M, "2023-06-30": 900.0 * _M},
            {"2023-03-31": "us-gaap:Cash", "2023-06-30": "us-gaap:Cash"},
        )
        _net_change_from_balances(rows, {"2023-03-31", "2023-06-30"}, "annual")
        assert _by_tag(rows, "net_change_in_cash").values == {}

    def test_different_tags_not_differenced(self):
        rows = self._rows(
            {"2022-12-31": 1000.0 * _M, "2023-12-31": 1200.0 * _M},
            {"2022-12-31": "us-gaap:Cash", "2023-12-31": "us-gaap:CashAndDue"},
        )
        _net_change_from_balances(rows, {"2022-12-31", "2023-12-31"}, "annual")
        assert _by_tag(rows, "net_change_in_cash").values == {}

    def test_reported_net_change_kept(self):
        rows = self._rows(
            {"2022-12-31": 1000.0 * _M, "2023-12-31": 1200.0 * _M},
            {"2022-12-31": "us-gaap:Cash", "2023-12-31": "us-gaap:Cash"},
            ncc={"2023-12-31": 150.0 * _M},
        )
        _net_change_from_balances(rows, {"2022-12-31", "2023-12-31"}, "annual")
        assert _by_tag(rows, "net_change_in_cash").values == {"2023-12-31": 150.0 * _M}

    def test_missing_rows_is_noop(self):
        rows = [_rr("cash_at_end_of_period", {_D: 1.0 * _M}, period_type="instant")]
        _net_change_from_balances(rows, {_D}, "annual")
        assert rows[0].values == {_D: 1.0 * _M}


class TestImputeTaxSign:
    def _rows(self, tax, current, deferred, current_source=None):
        return [
            _rr(
                "total_pretax_income",
                {_D: 1078.0 * _M},
                sequence=1,
                sources={
                    _D: "us-gaap:IncomeLossFromContinuingOperationsBeforeIncomeTaxes"
                },
            ),
            _rr(
                "income_tax_expense",
                {_D: tax * _M},
                sequence=2,
                sources={_D: "us-gaap:IncomeTaxExpenseBenefit"},
            ),
            _rr(
                "income_tax_current",
                {} if current is None else {_D: current * _M},
                sequence=3,
                sources={}
                if current is None
                else {_D: current_source or "us-gaap:CurrentIncomeTaxExpenseBenefit"},
            ),
            _rr(
                "income_tax_deferred",
                {_D: deferred * _M},
                sequence=4,
                sources={_D: "us-gaap:DeferredIncomeTaxExpenseBenefit"},
            ),
            _rr(
                "net_income_continuing",
                {_D: 769.0 * _M},
                sequence=5,
                sources={_D: "us-gaap:ProfitLoss"},
            ),
        ]

    def test_reversed_tax_sign_corrected(self):
        out, _ = impute(
            self._rows(-309.0, 323.0, -14.0),
            "income_statement",
            "industrial",
            {_D},
            facts={},
        )
        tax = _by_tag(out, "income_tax_expense")
        assert tax.values[_D] == 309.0 * _M
        assert tax.sources[_D] == (
            "corrected: total_pretax_income - net_income_continuing"
        )
        assert _by_tag(out, "total_pretax_income").values[_D] == 1078.0 * _M

    def test_imputed_current_tax_rederived_after_sign_correction(self):
        out, _ = impute(
            self._rows(-309.0, None, -14.0),
            "income_statement",
            "industrial",
            {_D},
            facts={},
        )
        assert _by_tag(out, "income_tax_expense").values[_D] == 309.0 * _M
        current = _by_tag(out, "income_tax_current")
        assert current.values[_D] == 323.0 * _M
        assert current.sources[_D] == (
            "imputed: income_tax_expense - income_tax_deferred"
        )

    def test_tax_sign_kept_when_components_agree(self):
        out, _ = impute(
            self._rows(-309.0, -295.0, -14.0),
            "income_statement",
            "industrial",
            {_D},
            facts={},
        )
        assert _by_tag(out, "income_tax_expense").values[_D] == -309.0 * _M


class TestImputeISCorrectionPasses:
    def test_cogs_kept_when_operating_expenses_rolled_up(self):
        rows = [
            _rr(
                "total_revenue",
                {_D: 1000.0 * _M},
                sequence=1,
                sources={_D: "us-gaap:Revenues"},
            ),
            _rr(
                "total_cost_of_revenue",
                {_D: 300.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:CostOfRevenue"},
            ),
            _rr(
                "costs_and_expenses",
                {_D: 900.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:CostsAndExpenses"},
            ),
            _rr("total_operating_expenses", {}, sequence=4),
            _rr(
                "sga_expense",
                {_D: 400.0 * _M},
                parent="total_operating_expenses",
                sequence=5,
                sources={_D: "us-gaap:SellingGeneralAndAdministrativeExpense"},
            ),
            _rr(
                "total_operating_income",
                {_D: 100.0 * _M},
                sequence=6,
                sources={_D: "us-gaap:OperatingIncomeLoss"},
            ),
        ]
        out, _ = impute(rows, "income_statement", "industrial", {_D}, facts={})
        cogs = _by_tag(out, "total_cost_of_revenue")
        assert cogs.values[_D] == 300.0 * _M
        assert cogs.sources[_D] == "us-gaap:CostOfRevenue"

    def test_gross_profit_rollup_recomputed_from_rev_minus_cogs(self):
        # gp carries an imputed-rollup source and a stale value; with cogs!=0 and
        # rev present it is recomputed to rev - cogs (407-414).
        rows = [
            _rr(
                "total_revenue",
                {_D: 1000.0 * _M},
                sequence=1,
                sources={_D: "us-gaap:Revenues"},
            ),
            _rr(
                "total_cost_of_revenue",
                {_D: 300.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:CostOfRevenue"},
            ),
            _rr(
                "total_gross_profit",
                {_D: 123.0 * _M},
                sequence=3,
                sources={_D: "imputed-rollup: segment_gp(+)"},
            ),
        ]
        out, _ = impute(rows, "income_statement", "industrial", {_D}, facts={})
        gp = _by_tag(out, "total_gross_profit")
        assert gp.values[_D] == 700.0 * _M  # 1000 - 300
        assert "imputed: total_revenue - total_cost_of_revenue" in gp.sources[_D]

    def test_cogs_backsolved_from_rev_minus_gross_profit(self):
        # gp hard-sourced and rev - cogs - gp violates identity -> cogs corrected
        # to rev - gp (499-506).
        rows = [
            _rr(
                "total_revenue",
                {_D: 1000.0 * _M},
                sequence=1,
                sources={_D: "us-gaap:Revenues"},
            ),
            _rr(
                "total_cost_of_revenue",
                {_D: 100.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:CostOfRevenue"},
            ),
            _rr(
                "total_gross_profit",
                {_D: 700.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:GrossProfit"},
            ),
        ]
        out, _ = impute(rows, "income_statement", "industrial", {_D}, facts={})
        cogs = _by_tag(out, "total_cost_of_revenue")
        assert cogs.values[_D] == 300.0 * _M  # 1000 - 700
        assert "corrected: total_revenue - total_gross_profit" in cogs.sources[_D]

    def test_opex_backsolved_from_gross_profit_minus_operating_income(self):
        # opex > gp triggers the opex correction to gp - opinc (527-540); opinc must
        # be hard-sourced so the "imputed" guard does not block it.
        rows = [
            _rr(
                "total_revenue",
                {_D: 1000.0 * _M},
                sequence=1,
                sources={_D: "us-gaap:Revenues"},
            ),
            _rr(
                "total_gross_profit",
                {_D: 800.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:GrossProfit"},
            ),
            _rr(
                "total_operating_expenses",
                {_D: 950.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:OperatingExpenses"},
            ),
            _rr(
                "total_operating_income",
                {_D: 300.0 * _M},
                sequence=4,
                sources={_D: "us-gaap:OperatingIncomeLoss"},
            ),
        ]
        out, _ = impute(rows, "income_statement", "industrial", {_D}, facts={})
        opex = _by_tag(out, "total_operating_expenses")
        assert opex.values[_D] == 500.0 * _M  # 800 - 300
        assert (
            "corrected: total_gross_profit - total_operating_income" in opex.sources[_D]
        )

    def test_opex_correction_skips_when_gross_profit_unresolvable(self):
        # No revenue/cogs to derive gp, opinc explicitly imputed -> in the opex
        # correction loop gp_val stays None so the loop `continue`s (524-525).
        rows = [
            _rr("total_gross_profit", {}, sequence=1),
            _rr(
                "total_operating_expenses",
                {_D: 950.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:OperatingExpenses"},
            ),
            _rr(
                "total_operating_income",
                {_D: 300.0 * _M},
                sequence=3,
                sources={_D: "imputed: x - y"},
            ),
        ]
        out, _ = impute(rows, "income_statement", "industrial", {_D}, facts={})
        opex = _by_tag(out, "total_operating_expenses")
        # gp never resolved -> correction loop skipped opex (no "corrected" marker).
        assert "corrected: total_gross_profit" not in opex.sources.get(_D, "")
        assert opex.values[_D] == 950.0 * _M


class TestImputeBSEquityReconcileGuard:
    def test_reconcile_skips_when_le_identity_violated(self):
        # ENCI != equity + nci (gap present) but L + ENCI + rNCI != L&E -> the
        # second guard `continue` (565-566) blocks the reconciliation.
        rows = [
            _rr(
                "total_assets",
                {_D: 1000.0 * _M},
                period_type="instant",
                balance="debit",
                sequence=1,
                sources={_D: "us-gaap:Assets"},
            ),
            _rr(
                "total_liabilities",
                {_D: 600.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=2,
                sources={_D: "us-gaap:Liabilities"},
            ),
            # L&E deliberately inconsistent with L + ENCI so the second guard fails.
            _rr(
                "total_liabilities_and_equity",
                {_D: 1000.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=3,
                sources={_D: "us-gaap:LiabilitiesAndStockholdersEquity"},
            ),
            _rr(
                "total_equity_and_noncontrolling_interests",
                {_D: 200.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=4,
                sources={
                    _D: "us-gaap:StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest"
                },
            ),
            _rr(
                "total_equity",
                {_D: 999.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=5,
                sources={_D: "us-gaap:StockholdersEquity"},
            ),
            _rr(
                "noncontrolling_interests",
                {_D: 0.0},
                period_type="instant",
                balance="credit",
                sequence=6,
                sources={_D: "us-gaap:MinorityInterest"},
            ),
        ]
        out, _ = impute(
            rows, "balance_sheet", "industrial", {_D}, facts={"us-gaap": {}}
        )
        eq = _by_tag(out, "total_equity")
        # Reconciliation did NOT fire (would have set 200): equity keeps its own value.
        assert "reconciled" not in eq.sources.get(_D, "")


# ---------------------------------------------------------------------------
# impute() -- cash-flow verify FX-scope and discontinued-ops fallbacks
# ---------------------------------------------------------------------------


def _cf_base(nc_val, nc_src):
    """Five CF rows whose op+inv+fin+fx = 200M, with a mismatching net_change."""
    return [
        _rr(
            "net_cash_from_operating_activities",
            {_D: 500.0 * _M},
            balance="debit",
            sequence=1,
            sources={_D: "us-gaap:NetCashProvidedByUsedInOperatingActivities"},
        ),
        _rr(
            "net_cash_from_investing_activities",
            {_D: -200.0 * _M},
            balance="debit",
            sequence=2,
            sources={_D: "us-gaap:NetCashProvidedByUsedInInvestingActivities"},
        ),
        _rr(
            "net_cash_from_financing_activities",
            {_D: -150.0 * _M},
            balance="debit",
            sequence=3,
            sources={_D: "us-gaap:NetCashProvidedByUsedInFinancingActivities"},
        ),
        _rr(
            "effect_of_exchange_rate_changes",
            {_D: 50.0 * _M},
            balance="debit",
            sequence=4,
            sources={_D: "us-gaap:EffectOfExchangeRateOnCash"},
        ),
        _rr(
            "net_change_in_cash",
            {_D: nc_val},
            balance="debit",
            sequence=5,
            sources={_D: nc_src},
        ),
    ]


class TestImputeCashFlowFXScope:
    def test_excluding_fx_marker_uses_no_fx_rule(self):
        # net_change marked ExcludingExchangeRateEffect -> the FX-inclusive rule is
        # skipped (640, 651); the no-FX rule (op+inv+fin=150) verifies cleanly.
        rows = _cf_base(
            150.0 * _M, "us-gaap:CashPeriodIncreaseDecreaseExcludingExchangeRateEffect"
        )
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts={"us-gaap": {}})
        # 500-200-150 = 150 matches the Excluding-FX net change -> verified, no warning.
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_including_fx_marker_skips_no_fx_rule(self):
        # net_change marked IncludingExchangeRateEffect with NO effect-of-fx row:
        # the FX rules cannot evaluate (source missing) and every no-FX rule is
        # skipped by the Including-scope guard (648) -> no diagnostic emitted.
        rows = _cf_base(
            200.0 * _M, "us-gaap:CashPeriodIncreaseDecreaseIncludingExchangeRateEffect"
        )
        rows = [r for r in rows if r.tag != "effect_of_exchange_rate_changes"]
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts={"us-gaap": {}})
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_continuing_operations_source_marks_scope_mismatch(self):
        # An activity source carries ContinuingOperations while net_change is FX-
        # scoped -> _cf_scope_mismatch=True (666-667); the engine scope-aligns the
        # net change rather than warning.
        rows = _cf_base(
            999.0 * _M, "us-gaap:CashPeriodIncreaseDecreaseExcludingExchangeRateEffect"
        )
        rows[0].sources[_D] = (
            "us-gaap:NetCashProvidedByUsedInOperatingActivitiesContinuingOperations"
        )
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts={"us-gaap": {}})
        nc = _by_tag(out, "net_change_in_cash")
        assert all(w.tag != "net_change_in_cash" for w in diag)
        assert "scope-aligned" in nc.sources[_D]


class TestImputeCashFlowDiscFallbacks:
    def test_disc_row_with_disc_fx_fact_closes_gap(self):
        # An explicit disc-ops row plus a discontinued-ops FX fact close the gap
        # via the disc-FX adjustment (984-1006).
        facts = {
            "us-gaap": {
                "EffectOfExchangeRateOnCashAndCashEquivalentsDiscontinuedOperations": {
                    "units": {"USD": [_dur(_D, "2023-01-01", 10.0 * _M)]}
                }
            }
        }
        rows = _cf_base(240.0 * _M, "us-gaap:CashIncludingExchangeRateEffect")
        rows.append(
            _rr(
                "net_cash_from_discontinued_operations",
                {_D: 30.0 * _M},
                balance="debit",
                sequence=6,
                sources={_D: "us-gaap:NetCashProvidedByUsedInDiscontinuedOperations"},
            )
        )
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        # 200 + 30 (disc) + 10 (disc-fx) = 240 matches.
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_disc_row_fallback_without_disc_fx(self):
        # disc row present, disc-fx present but the with-fx identity misses while the
        # without-fx identity matches (1007-1013).
        facts = {
            "us-gaap": {
                "EffectOfExchangeRateOnCashAndCashEquivalentsDiscontinuedOperations": {
                    "units": {"USD": [_dur(_D, "2023-01-01", 10.0 * _M)]}
                }
            }
        }
        rows = _cf_base(230.0 * _M, "us-gaap:CashIncludingExchangeRateEffect")
        rows.append(
            _rr(
                "net_cash_from_discontinued_operations",
                {_D: 30.0 * _M},
                balance="debit",
                sequence=6,
                sources={_D: "us-gaap:NetCashProvidedByUsedInDiscontinuedOperations"},
            )
        )
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        # 200 + 30 = 230 (without disc-fx) matches; with disc-fx (240) would not.
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_disc_ops_sum_fallback_closes_gap(self):
        # No disc row; summed individual disc-ops activity tags reconcile (1051-1122).
        facts = {
            "us-gaap": {
                "NetCashProvidedByUsedInDiscontinuedOperations": {
                    "units": {"USD": [_dur(_D, "2023-01-01", 30.0 * _M)]}
                }
            }
        }
        rows = _cf_base(230.0 * _M, "us-gaap:CashIncludingExchangeRateEffect")
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_disc_ops_sum_fallback_skips_malformed_start(self):
        # The disc-ops-sum fallback (1051-1122) iterates the discontinued NetCash tag;
        # a malformed-dated entry is skipped (1082-1083) before the clean entry sums in.
        facts = {
            "us-gaap": {
                "NetCashProvidedByUsedInDiscontinuedOperations": {
                    "units": {
                        "USD": [
                            {
                                "end": _D,
                                "start": "bad-date",
                                "val": 99.0 * _M,
                                "form": "10-K",
                                "filed": "2024-01-15",
                            },  # 1082-1083
                            _dur(_D, "2023-01-01", 30.0 * _M),  # clean -> sums to 30
                        ]
                    }
                }
            }
        }
        rows = _cf_base(230.0 * _M, "us-gaap:CashIncludingExchangeRateEffect")
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        # 200 (op+inv+fin+fx) + 30 (disc sum) = 230 -> reconciled.
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_disc_individual_tag_skips_malformed_start(self):
        # The individual disc-ops fallback (1020-1049) skips a malformed-dated entry
        # (1033-1034) then matches the clean one to close the gap.
        facts = {
            "us-gaap": {
                "CashProvidedByUsedInOperatingActivitiesDiscontinuedOperations": {
                    "units": {
                        "USD": [
                            {
                                "end": _D,
                                "start": "nope",
                                "val": 99.0 * _M,
                                "form": "10-K",
                                "filed": "2024-01-15",
                            },  # 1033-1034
                            _dur(
                                _D, "2023-01-01", 30.0 * _M
                            ),  # clean -> 200 + 30 = 230
                        ]
                    }
                }
            }
        }
        rows = _cf_base(230.0 * _M, "us-gaap:CashIncludingExchangeRateEffect")
        # An explicit disc row routes through the individual-tags fallback (disc_val set).
        rows.append(
            _rr(
                "net_cash_from_discontinued_operations",
                {_D: 0.0},
                balance="debit",
                sequence=6,
                sources={_D: "us-gaap:NetCashProvidedByUsedInDiscontinuedOperations"},
            )
        )
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_disc_row_disc_fx_skips_malformed_start(self):
        # disc row present: the disc-FX adjustment loop (982-998) skips a malformed-
        # dated FX entry (990-991) then uses the clean one to close the gap.
        facts = {
            "us-gaap": {
                "EffectOfExchangeRateOnCashAndCashEquivalentsDiscontinuedOperations": {
                    "units": {
                        "USD": [
                            {
                                "end": _D,
                                "start": "xx",
                                "val": 99.0 * _M,
                                "form": "10-K",
                                "filed": "2024-01-15",
                            },  # 990-991
                            _dur(_D, "2023-01-01", 10.0 * _M),  # clean
                        ]
                    }
                }
            }
        }
        rows = _cf_base(240.0 * _M, "us-gaap:CashIncludingExchangeRateEffect")
        rows.append(
            _rr(
                "net_cash_from_discontinued_operations",
                {_D: 30.0 * _M},
                balance="debit",
                sequence=6,
                sources={_D: "us-gaap:NetCashProvidedByUsedInDiscontinuedOperations"},
            )
        )
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        # 200 + 30 (disc) + 10 (disc-fx) = 240.
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_disc_ops_sum_with_disc_fx_fallback(self):
        # No disc row: the disc-ops-sum fallback finds the discontinued sum then the
        # disc-FX-fb1 loop (1093-1122) skips a malformed FX entry (1104-1105) and adds
        # the clean disc-FX value to reconcile (1107-1112).
        facts = {
            "us-gaap": {
                "NetCashProvidedByUsedInDiscontinuedOperations": {
                    "units": {"USD": [_dur(_D, "2023-01-01", 30.0 * _M)]}
                },
                "EffectOfExchangeRateOnCashAndCashEquivalentsDiscontinuedOperations": {
                    "units": {
                        "USD": [
                            {
                                "end": _D,
                                "start": "zz",
                                "val": 99.0 * _M,
                                "form": "10-K",
                                "filed": "2024-01-15",
                            },  # 1104-1105
                            _dur(
                                _D, "2023-01-01", 10.0 * _M
                            ),  # clean -> 200+30+10 = 240
                        ]
                    }
                },
            }
        }
        rows = _cf_base(240.0 * _M, "us-gaap:CashIncludingExchangeRateEffect")
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_disposal_group_change_closes_gap(self):
        # A disposal-group net-change fact equal to the activity sum (val) bridges
        # the gap: the check reduces to val ~= disposal_nc (1124-1173).
        facts = {
            "us-gaap": {
                "CashAndCashEquivalentsPeriodIncreaseDecreaseDisposalGroupIncludingDiscontinuedOperations": {
                    "units": {"USD": [_dur(_D, "2023-01-01", 200.0 * _M)]}
                }
            }
        }
        rows = _cf_base(230.0 * _M, "us-gaap:CashIncludingExchangeRateEffect")
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_disposal_cash_balance_delta_closes_gap(self):
        # Instant disposal-group cash balances (end vs prior quarter) reconcile via
        # the cash-delta path: val - (end - start) ~= net_change (1178-1224).
        prior = "2023-09-30"  # prior_period_end("2023-12-31")
        facts = {
            "us-gaap": {
                "DisposalGroupIncludingDiscontinuedOperationCashAndCashEquivalents": {
                    "units": {"USD": [_inst(_D, 80.0 * _M), _inst(prior, 50.0 * _M)]}
                }
            }
        }
        # delta = 80 - 50 = 30; identity uses val - delta -> 200 - 30 = 170 target.
        rows = _cf_base(170.0 * _M, "us-gaap:CashIncludingExchangeRateEffect")
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_disposal_group_change_skips_malformed_start(self):
        # The disposal-group net-change loop (1137-1173) skips a malformed-dated entry
        # (1153-1154) then bridges with the clean disposal fact.
        facts = {
            "us-gaap": {
                "CashAndCashEquivalentsPeriodIncreaseDecreaseDisposalGroupIncludingDiscontinuedOperations": {
                    "units": {
                        "USD": [
                            {
                                "end": _D,
                                "start": "??",
                                "val": 99.0 * _M,
                                "form": "10-K",
                                "filed": "2024-01-15",
                            },  # 1153-1154
                            _dur(
                                _D, "2023-01-01", 200.0 * _M
                            ),  # clean: check reduces to val ~= 200
                        ]
                    }
                }
            }
        }
        rows = _cf_base(230.0 * _M, "us-gaap:CashIncludingExchangeRateEffect")
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_restricted_cash_change_skips_malformed_start(self):
        # No disc/disposal facts: the loop reaches the restricted-cash block (1226-1262)
        # which skips a malformed-dated entry (1244-1245) then adds the clean change.
        facts = {
            "us-gaap": {
                "IncreaseDecreaseInRestrictedCashAndRestrictedCashEquivalents": {
                    "units": {
                        "USD": [
                            {
                                "end": _D,
                                "start": "!!",
                                "val": 99.0 * _M,
                                "form": "10-K",
                                "filed": "2024-01-15",
                            },  # 1244-1245
                            _dur(
                                _D, "2023-01-01", 30.0 * _M
                            ),  # clean -> 200 + 30 = 230
                        ]
                    }
                }
            }
        }
        rows = _cf_base(230.0 * _M, "us-gaap:CashIncludingExchangeRateEffect")
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_activity_pairs_scope_skips_malformed_start(self):
        # The continuing-vs-total activity-pairs scope block (1264-1383) reads activity
        # facts via _cf_vals, which skips a malformed-dated entry (1325-1326) then uses
        # the clean continuing-operations values to reconcile.
        def _pair(tot_val):
            return {
                "units": {
                    "USD": [
                        {
                            "end": _D,
                            "start": "##",
                            "val": 99.0 * _M,
                            "form": "10-K",
                            "filed": "2024-01-15",
                        },  # 1325-1326
                        _dur(_D, "2023-01-01", tot_val),  # clean
                    ]
                }
            }

        facts = {
            "us-gaap": {
                "NetCashProvidedByUsedInOperatingActivitiesContinuingOperations": _pair(
                    500.0 * _M
                ),
                "NetCashProvidedByUsedInInvestingActivitiesContinuingOperations": _pair(
                    -200.0 * _M
                ),
                "NetCashProvidedByUsedInFinancingActivitiesContinuingOperations": _pair(
                    -150.0 * _M
                ),
                # A discontinued-ops FX fact triggers the disc-FX combination loop
                # (1356-1359) that augments the FX option set.
                "EffectOfExchangeRateOnCashAndCashEquivalentsDiscontinuedOperations": {
                    "units": {"USD": [_dur(_D, "2023-01-01", 20.0 * _M)]}
                },
                "CashCashEquivalentsRestrictedCashAndRestrictedCashEquivalentsPeriodIncreaseDecreaseIncludingExchangeRateEffect": {
                    "units": {"USD": [_dur(_D, "2023-01-01", 150.0 * _M)]}
                },
            }
        }
        # Activity opts 500/-200/-150 + fx(0) reconcile to the 150 net-change fact.
        rows = _cf_base(150.0 * _M, "us-gaap:CashIncludingExchangeRateEffect")
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_cash_balance_delta_reconciles_net_change(self):
        # No disc/disposal/activity facts: the loop falls through to the cash-balance
        # delta block (1385-1438).  The instant CashCashEquivalents... facts include a
        # duration entry (skipped, 1409) and a malformed-dated instant (1421-1422)
        # alongside the clean end/prior instants whose delta equals net_change.
        prior = "2023-06-30"  # 184 days before _D -> inside the 60..400 window
        facts = {
            "us-gaap": {
                "CashCashEquivalentsRestrictedCashAndRestrictedCashEquivalents": {
                    "units": {
                        "USD": [
                            _dur(
                                _D, "2023-01-01", 999.0 * _M
                            ),  # has start -> 1409 skip
                            _inst(_D, 380.0 * _M),  # end-of-period balance
                            _inst(prior, 80.0 * _M),  # prior-period balance
                            {
                                "end": "garbage",
                                "val": 7.0 * _M,
                                "form": "10-K",
                                "filed": "2024-02-15",
                            },  # 1421-1422
                        ]
                    }
                }
            }
        }
        # delta = 380 - 80 = 300 == net_change; the FX rule (op+inv+fin+fx=200) misses
        # so the loop reaches the balance-delta reconciliation.
        rows = _cf_base(300.0 * _M, "us-gaap:CashIncludingExchangeRateEffect")
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        assert all(w.tag != "net_change_in_cash" for w in diag)


# ---------------------------------------------------------------------------
# impute() -- income-statement verify-loop fallbacks
# ---------------------------------------------------------------------------


class TestImputeISVerifyFallbacks:
    def test_disc_adjusted_nic_suppresses_pretax_warning(self):
        # nic becomes ProfitLoss "(disc-adjusted)" and tax is not ContinuingOps; with
        # pretax != nic + tax the verify loop short-circuits both pairs (683-698)
        # rather than warning.
        # tax row carries a value but an empty source: the pre-pass applies the disc
        # adjustment (empty tax-src is not a ContinuingOps mismatch), and the verify
        # loop then sees a non-ContinuingOps tax -> short-circuits both pairs.
        rows = [
            _rr(
                "total_pretax_income",
                {_D: 900.0 * _M},
                sequence=1,
                sources={_D: "us-gaap:IncomeBeforeTax"},
            ),
            _rr("income_tax_expense", {_D: 100.0 * _M}, sequence=2),
            _rr(
                "net_income_continuing",
                {_D: 430.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:ProfitLoss"},
            ),
            _rr("net_income_discontinued", {_D: 30.0 * _M}, sequence=4),
        ]
        out, diag = impute(
            rows, "income_statement", "industrial", {_D}, facts={"us-gaap": {}}
        )
        nic = _by_tag(out, "net_income_continuing")
        assert "(disc-adjusted)" in nic.sources[_D]
        # No pretax/nic identity warning despite 900 != 400 + 100.
        assert all(
            w.tag not in ("total_pretax_income", "net_income_continuing") for w in diag
        )

    def test_profitloss_swap_from_facts_resolves_identity(self):
        # nic from ProfitLoss; a NetIncomeLoss fact (parent-only) makes pretax=nic+tax
        # hold -> swap accepted (827-840) without warning. Mirrors but isolates the
        # us-gaap NetIncomeLoss lookup.
        facts = {
            "us-gaap": {
                "NetIncomeLoss": {
                    "units": {"USD": [_dur(_D, "2023-01-01", 550.0 * _M)]}
                }
            }
        }
        rows = [
            _rr(
                "total_pretax_income",
                {_D: 700.0 * _M},
                sequence=1,
                sources={
                    _D: "us-gaap:IncomeLossFromContinuingOperationsBeforeIncomeTaxes"
                },
            ),
            _rr(
                "income_tax_expense",
                {_D: 150.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:IncomeTaxExpenseBenefit"},
            ),
            _rr(
                "net_income_continuing",
                {_D: 600.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:ProfitLoss"},
            ),
        ]
        out, diag = impute(rows, "income_statement", "diversified", {_D}, facts=facts)
        nic = _by_tag(out, "net_income_continuing")
        assert nic.values[_D] == 550.0 * _M
        assert "NCI-corrected" in nic.sources[_D]

    def test_profitloss_swap_for_netincomeloss_nci_pretax(self):
        # nic from NetIncomeLoss, pretax tagged with NoncontrollingInterest -> the
        # engine swaps in a ProfitLoss fact that satisfies pretax=nic+tax (841-860).
        facts = {
            "us-gaap": {
                "ProfitLoss": {"units": {"USD": [_dur(_D, "2023-01-01", 550.0 * _M)]}}
            }
        }
        rows = [
            _rr(
                "total_pretax_income",
                {_D: 700.0 * _M},
                sequence=1,
                # Carries NoncontrollingInterest but NOT EquityMethodInvestments, so
                # the equity-method pre-pass leaves pretax intact.
                sources={
                    _D: "us-gaap:IncomeLossFromContinuingOperationsBeforeIncomeTaxesNoncontrollingInterest"
                },
            ),
            _rr(
                "income_tax_expense",
                {_D: 150.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:IncomeTaxExpenseBenefit"},
            ),
            _rr(
                "net_income_continuing",
                {_D: 600.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:NetIncomeLoss"},
            ),
        ]
        out, diag = impute(rows, "income_statement", "diversified", {_D}, facts=facts)
        nic = _by_tag(out, "net_income_continuing")
        assert nic.values[_D] == 550.0 * _M
        assert "ProfitLoss(NCI-corrected)" in nic.sources[_D]

    def test_q4_nci_swap_from_fy_minus_quarters(self):
        # nic is Q4-derived; an annual NetIncomeLoss plus 3 quarterly NetIncomeLoss
        # facts reconstruct a parent-only Q4 that satisfies the identity (862-952).
        q_ends = ["2023-03-31", "2023-06-30", "2023-09-30"]
        nil_entries = [
            _dur(_D, "2023-01-01", 600.0 * _M),  # FY parent-only NI
            _dur(q_ends[0], "2023-01-01", 100.0 * _M),
            _dur(q_ends[1], "2023-04-01", 150.0 * _M),
            _dur(q_ends[2], "2023-07-01", 200.0 * _M),
        ]
        facts = {"us-gaap": {"NetIncomeLoss": {"units": {"USD": nil_entries}}}}
        # parent-only Q4 = 600 - (100+150+200) = 150; identity: pretax(250)=150+tax(100).
        rows = [
            _rr(
                "total_pretax_income",
                {_D: 250.0 * _M},
                sequence=1,
                sources={
                    _D: "us-gaap:IncomeLossFromContinuingOperationsBeforeIncomeTaxes"
                },
            ),
            _rr(
                "income_tax_expense",
                {_D: 100.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:IncomeTaxExpenseBenefit"},
            ),
            _rr(
                "net_income_continuing",
                {_D: 999.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:NetIncomeLoss Q4: FY[..] - (..)"},
            ),
        ]
        out, diag = impute(rows, "income_statement", "diversified", {_D}, facts=facts)
        nic = _by_tag(out, "net_income_continuing")
        assert nic.values[_D] == 150.0 * _M
        assert "Q4-NCI-corrected" in nic.sources[_D]

    def test_pick_entries_filters_bad_entries_and_rejects_off_identity(self):
        # Exercises _pick_entries' skip paths and _try_nci_swap's reject path via the
        # ProfitLoss NCI-swap branch.  The NetIncomeLoss facts include entries that are
        # skipped (wrong end / no start / wrong form -> 774), malformed-dated (781-782),
        # out-of-window (789), an off-identity but well-formed candidate (filed first ->
        # _try_nci_swap returns False, 825), then a matching candidate that swaps.
        nil_entries = [
            {
                "end": "2022-12-31",
                "start": "2022-01-01",
                "val": 9.0 * _M,
                "form": "10-K",
                "filed": "2023-01-01",
            },  # wrong end -> 774
            {
                "end": _D,
                "val": 9.0 * _M,
                "form": "10-K",
                "filed": "2023-02-01",
            },  # no start -> 774
            {
                "end": _D,
                "start": "2023-01-01",
                "val": 9.0 * _M,
                "form": "8-K",
                "filed": "2023-03-01",
            },  # bad form -> 774
            {
                "end": _D,
                "start": "not-a-date",
                "val": 9.0 * _M,
                "form": "10-K",
                "filed": "2023-04-01",
            },  # malformed -> 781-782
            {
                "end": _D,
                "start": "2023-12-20",
                "val": 9.0 * _M,
                "form": "10-K",
                "filed": "2023-05-01",
            },  # 11 days -> 789
            _dur(
                _D, "2023-01-01", 500.0 * _M, filed="2024-01-15"
            ),  # well-formed, off-identity -> 825 False
            _dur(
                _D, "2023-01-01", 550.0 * _M, filed="2024-02-15"
            ),  # matches identity -> swap
        ]
        facts = {"us-gaap": {"NetIncomeLoss": {"units": {"USD": nil_entries}}}}
        rows = [
            _rr(
                "total_pretax_income",
                {_D: 700.0 * _M},
                sequence=1,
                sources={
                    _D: "us-gaap:IncomeLossFromContinuingOperationsBeforeIncomeTaxes"
                },
            ),
            _rr(
                "income_tax_expense",
                {_D: 150.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:IncomeTaxExpenseBenefit"},
            ),
            _rr(
                "net_income_continuing",
                {_D: 600.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:ProfitLoss"},
            ),
        ]
        out, _ = impute(rows, "income_statement", "diversified", {_D}, facts=facts)
        nic = _by_tag(out, "net_income_continuing")
        assert nic.values[_D] == 550.0 * _M  # the matching candidate, not 500
        assert "NCI-corrected" in nic.sources[_D]

    def test_q4_nci_swap_skips_malformed_and_out_of_window_quarters(self):
        # Q4-derived nic: the FY NetIncomeLoss scan and the quarterly scan each skip
        # malformed-dated / out-of-window / wrong-form entries (893-894, 917, 924-925,
        # 933) before reconstructing a clean parent-only Q4 that satisfies the identity.
        nil_entries = [
            {
                "end": _D,
                "start": "bad",
                "val": 600.0 * _M,
                "form": "10-K",
                "filed": "2024-01-10",
            },  # FY malformed -> 893-894
            _dur(
                _D, "2023-01-01", 600.0 * _M, filed="2024-02-10"
            ),  # clean FY parent-only
            {
                "end": "2023-03-31",
                "start": "2023-01-01",
                "val": 100.0 * _M,
                "form": "8-K",
                "filed": "2023-04-10",
            },  # wrong form -> 917
            {
                "end": "2023-06-30",
                "start": "rotten",
                "val": 150.0 * _M,
                "form": "10-Q",
                "filed": "2023-07-10",
            },  # malformed -> 924-925
            {
                "end": "2024-06-30",
                "start": "2024-04-01",
                "val": 999.0 * _M,
                "form": "10-Q",
                "filed": "2024-07-10",
            },  # end >= date -> 933
            _dur(
                "2023-03-31", "2023-01-01", 100.0 * _M, form="10-Q", filed="2023-04-20"
            ),
            _dur(
                "2023-06-30", "2023-04-01", 150.0 * _M, form="10-Q", filed="2023-07-20"
            ),
            _dur(
                "2023-09-30", "2023-07-01", 200.0 * _M, form="10-Q", filed="2023-10-20"
            ),
        ]
        facts = {"us-gaap": {"NetIncomeLoss": {"units": {"USD": nil_entries}}}}
        # parent-only Q4 = 600 - (100+150+200) = 150; identity pretax(250)=150+tax(100).
        rows = [
            _rr(
                "total_pretax_income",
                {_D: 250.0 * _M},
                sequence=1,
                sources={
                    _D: "us-gaap:IncomeLossFromContinuingOperationsBeforeIncomeTaxes"
                },
            ),
            _rr(
                "income_tax_expense",
                {_D: 100.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:IncomeTaxExpenseBenefit"},
            ),
            _rr(
                "net_income_continuing",
                {_D: 999.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:NetIncomeLoss Q4: FY[..] - (..)"},
            ),
        ]
        out, _ = impute(rows, "income_statement", "diversified", {_D}, facts=facts)
        nic = _by_tag(out, "net_income_continuing")
        assert nic.values[_D] == 150.0 * _M
        assert "Q4-NCI-corrected" in nic.sources[_D]

    def test_q4_nci_swap_skips_alt_tag_without_fy_value(self):
        # Q4-derived nic: the NetIncomeLoss alt-tag has no full-year entry so its scan
        # yields no FY value and the loop advances to the next alt-tag (907-908); the
        # ProfitLoss alt-tag then carries the full quarterly+FY set that reconstructs Q4.
        q_ends = ["2023-03-31", "2023-06-30", "2023-09-30"]
        pl_entries = [
            _dur(_D, "2023-01-01", 600.0 * _M),  # FY parent-only NI on ProfitLoss
            _dur(q_ends[0], "2023-01-01", 100.0 * _M),
            _dur(q_ends[1], "2023-04-01", 150.0 * _M),
            _dur(q_ends[2], "2023-07-01", 200.0 * _M),
        ]
        facts = {
            "us-gaap": {
                # NetIncomeLoss only has quarter entries -> no 300..400 day FY value.
                "NetIncomeLoss": {
                    "units": {"USD": [_dur(q_ends[0], "2023-01-01", 100.0 * _M)]}
                },
                "ProfitLoss": {"units": {"USD": pl_entries}},
            }
        }
        rows = [
            _rr(
                "total_pretax_income",
                {_D: 250.0 * _M},
                sequence=1,
                sources={
                    _D: "us-gaap:IncomeLossFromContinuingOperationsBeforeIncomeTaxes"
                },
            ),
            _rr(
                "income_tax_expense",
                {_D: 100.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:IncomeTaxExpenseBenefit"},
            ),
            _rr(
                "net_income_continuing",
                {_D: 999.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:ProfitLoss Q4: FY[..] - (..)"},
            ),
        ]
        out, _ = impute(rows, "income_statement", "diversified", {_D}, facts=facts)
        nic = _by_tag(out, "net_income_continuing")
        assert nic.values[_D] == 150.0 * _M
        assert "Q4-NCI-corrected" in nic.sources[_D]

    def test_total_assets_sign_corrected_from_negative_le(self):
        # total_assets verifies against a negative total_liabilities_and_equity whose
        # magnitude matches -> L&E is sign-corrected (728-750).
        rows = [
            _rr(
                "total_assets",
                {_D: 1000.0 * _M},
                period_type="instant",
                balance="debit",
                sequence=1,
                sources={_D: "us-gaap:Assets"},
            ),
            _rr(
                "total_liabilities_and_equity",
                {_D: -1000.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=2,
                sources={_D: "us-gaap:LiabilitiesAndStockholdersEquity"},
            ),
        ]
        out, diag = impute(
            rows, "balance_sheet", "industrial", {_D}, facts={"us-gaap": {}}
        )
        le = _by_tag(out, "total_liabilities_and_equity")
        assert le.values[_D] == 1000.0 * _M
        assert "sign-corrected" in le.sources[_D]
        assert all(w.tag != "total_assets" for w in diag)


# ---------------------------------------------------------------------------
# impute() -- balance-sheet mezzanine / operating-income verify fallbacks
# ---------------------------------------------------------------------------


def _bs_mezz_rows(l_val, enci_val, rnci_val=None):
    rows = [
        _rr(
            "total_assets",
            {_D: 1000.0 * _M},
            period_type="instant",
            balance="debit",
            sequence=1,
            sources={_D: "us-gaap:Assets"},
        ),
        _rr(
            "total_liabilities",
            {_D: l_val},
            period_type="instant",
            balance="credit",
            sequence=2,
            sources={_D: "us-gaap:Liabilities"},
        ),
        _rr(
            "total_liabilities_and_equity",
            {_D: 1000.0 * _M},
            period_type="instant",
            balance="credit",
            sequence=3,
            sources={_D: "us-gaap:LiabilitiesAndStockholdersEquity"},
        ),
        _rr(
            "total_equity_and_noncontrolling_interests",
            {_D: enci_val},
            period_type="instant",
            balance="credit",
            sequence=4,
            sources={
                _D: "us-gaap:StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest"
            },
        ),
    ]
    if rnci_val is not None:
        rows.append(
            _rr(
                "redeemable_noncontrolling_interest",
                {_D: rnci_val},
                period_type="instant",
                balance="credit",
                sequence=5,
                sources={
                    _D: "us-gaap:RedeemableNoncontrollingInterestEquityCarryingAmount"
                },
            )
        )
    return rows


class TestImputeBSMezzanineFallbacks:
    def test_liabilities_zero_mezzanine_remainder_verified(self):
        # rule-with-rnci fails (rnci=50 makes val=550 != 600) but le - l - enci == 0
        # -> the near-zero mezzanine branch verifies (1466-1487).
        rows = _bs_mezz_rows(600.0 * _M, 400.0 * _M, rnci_val=50.0 * _M)
        # A mezzanine instant fact is present so the accumulation loop (1466-1468) runs.
        facts = {
            "us-gaap": {
                "TemporaryEquityCarryingAmount": {"units": {"USD": [_inst(_D, 0.0)]}}
            }
        }
        out, diag = impute(rows, "balance_sheet", "industrial", {_D}, facts=facts)
        assert all(w.tag != "total_liabilities" for w in diag)

    def test_equity_resolved_from_individual_mezzanine_fact(self):
        # No liabilities-and-equity total: the equity reconciliation does not apply.
        rows = [
            _rr(
                "total_assets",
                {_D: 1000.0 * _M},
                period_type="instant",
                balance="debit",
                sequence=1,
                sources={_D: "us-gaap:Assets"},
            ),
            _rr(
                "total_liabilities",
                {_D: 600.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=2,
                sources={_D: "us-gaap:Liabilities"},
            ),
            _rr(
                "total_equity_and_noncontrolling_interests",
                {_D: 400.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=4,
                sources={
                    _D: "us-gaap:StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest"
                },
            ),
            _rr(
                "total_equity",
                {_D: 350.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=5,
                sources={_D: "us-gaap:StockholdersEquity"},
            ),
            _rr(
                "noncontrolling_interests",
                {_D: 0.0},
                period_type="instant",
                balance="credit",
                sequence=6,
                sources={_D: "us-gaap:MinorityInterest"},
            ),
            _rr(
                "redeemable_noncontrolling_interest",
                {_D: 50.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=7,
                sources={
                    _D: "us-gaap:RedeemableNoncontrollingInterestEquityCarryingAmount"
                },
            ),
        ]
        facts = {
            "us-gaap": {
                "RedeemableNoncontrollingInterestEquityCarryingAmount": {
                    "units": {"USD": [_inst(_D, 50.0 * _M)]}
                }
            }
        }
        out, diag = impute(rows, "balance_sheet", "industrial", {_D}, facts=facts)
        assert all(w.tag != "total_equity_and_noncontrolling_interests" for w in diag)

    def test_equity_resolved_from_summed_mezzanine_facts(self):
        # No liabilities-and-equity total: the equity reconciliation does not apply.
        rows = [
            _rr(
                "total_assets",
                {_D: 1000.0 * _M},
                period_type="instant",
                balance="debit",
                sequence=1,
                sources={_D: "us-gaap:Assets"},
            ),
            _rr(
                "total_liabilities",
                {_D: 600.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=2,
                sources={_D: "us-gaap:Liabilities"},
            ),
            _rr(
                "total_equity_and_noncontrolling_interests",
                {_D: 400.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=4,
                sources={
                    _D: "us-gaap:StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest"
                },
            ),
            _rr(
                "total_equity",
                {_D: 350.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=5,
                sources={_D: "us-gaap:StockholdersEquity"},
            ),
            _rr(
                "noncontrolling_interests",
                {_D: 0.0},
                period_type="instant",
                balance="credit",
                sequence=6,
                sources={_D: "us-gaap:MinorityInterest"},
            ),
            _rr(
                "redeemable_noncontrolling_interest",
                {_D: 50.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=7,
                sources={
                    _D: "us-gaap:RedeemableNoncontrollingInterestEquityCarryingAmount"
                },
            ),
        ]
        facts = {
            "us-gaap": {
                "RedeemableNoncontrollingInterestEquityCommonCarryingAmount": {
                    "units": {"USD": [_inst(_D, 30.0 * _M)]}
                },
                "RedeemableNoncontrollingInterestEquityPreferredCarryingAmount": {
                    "units": {"USD": [_inst(_D, 20.0 * _M)]}
                },
            }
        }
        out, diag = impute(rows, "balance_sheet", "industrial", {_D}, facts=facts)
        assert all(w.tag != "total_equity_and_noncontrolling_interests" for w in diag)

    def test_equity_negative_nci_double_counted(self):
        # No liabilities-and-equity total: the equity reconciliation does not apply.
        rows = [
            _rr(
                "total_assets",
                {_D: 1000.0 * _M},
                period_type="instant",
                balance="debit",
                sequence=1,
                sources={_D: "us-gaap:Assets"},
            ),
            _rr(
                "total_liabilities",
                {_D: 600.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=2,
                sources={_D: "us-gaap:Liabilities"},
            ),
            # equity=420, nci=-20 -> verify sum = 400; ENCI value = 440 -> diff = 40 = 2*20.
            _rr(
                "total_equity_and_noncontrolling_interests",
                {_D: 440.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=4,
                sources={
                    _D: "us-gaap:StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest"
                },
            ),
            _rr(
                "total_equity",
                {_D: 420.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=5,
                sources={_D: "us-gaap:StockholdersEquity"},
            ),
            _rr(
                "noncontrolling_interests",
                {_D: -20.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=6,
                sources={_D: "us-gaap:MinorityInterest"},
            ),
        ]
        out, diag = impute(
            rows, "balance_sheet", "industrial", {_D}, facts={"us-gaap": {}}
        )
        assert all(w.tag != "total_equity_and_noncontrolling_interests" for w in diag)


class TestImputeOperatingIncomeBridge:
    def test_disposition_gain_bridges_operating_income(self):
        # gp - opex != opinc (opinc imputed-sourced so pre-correction is skipped); a
        # GainLossOnDispositionOfAssets fact equals the signed gap -> bridged (1560-1612).
        rows = [
            _rr(
                "total_revenue",
                {_D: 1000.0 * _M},
                sequence=1,
                sources={_D: "us-gaap:Revenues"},
            ),
            _rr(
                "total_gross_profit",
                {_D: 1000.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:GrossProfit"},
            ),
            _rr(
                "total_operating_expenses",
                {_D: 300.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:OperatingExpenses"},
            ),
            _rr(
                "total_operating_income",
                {_D: 730.0 * _M},
                sequence=4,
                sources={_D: "imputed: revenue - expenses"},
            ),
        ]
        facts = {
            "us-gaap": {
                "GainLossOnDispositionOfAssets": {
                    "units": {"USD": [_dur(_D, "2023-01-01", 30.0 * _M)]}
                }
            }
        }
        out, diag = impute(rows, "income_statement", "industrial", {_D}, facts=facts)
        assert all(w.tag != "total_operating_income" for w in diag)

    def test_operating_income_rounding_heuristic_bridges(self):
        # No bridging fact: gp/opex/opinc are all whole millions and the 1M residual
        # exceeds the scale tolerance (0.1% of 300M = 300k) yet is <= 1M -> the
        # rounding heuristic accepts it (1593-1608).
        rows = [
            _rr(
                "total_revenue",
                {_D: 400.0 * _M},
                sequence=1,
                sources={_D: "us-gaap:Revenues"},
            ),
            _rr(
                "total_gross_profit",
                {_D: 300.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:GrossProfit"},
            ),
            _rr(
                "total_operating_expenses",
                {_D: 100.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:OperatingExpenses"},
            ),
            # gp - opex = 200M; opinc 201M -> 1M residual; opinc imputed so the opex
            # pre-correction pass is skipped and the residual survives into verify.
            _rr(
                "total_operating_income",
                {_D: 201.0 * _M},
                sequence=4,
                sources={_D: "imputed: revenue - expenses"},
            ),
        ]
        out, diag = impute(
            rows, "income_statement", "industrial", {_D}, facts={"us-gaap": {}}
        )
        assert all(w.tag != "total_operating_income" for w in diag)


# ---------------------------------------------------------------------------
# Generic identity-enforcement tail (soft sources, vintage, derived markers)
# ---------------------------------------------------------------------------


def _bs_assets_rows(*, assets_src, tle_src, assets=1000.0, tle=900.0):
    """total_assets vs total_liabilities_and_equity (single-source BS_VERIFY rule)."""
    return [
        _rr(
            "total_assets",
            {_D: assets * _M},
            balance="debit",
            sequence=1,
            sources={_D: assets_src},
        ),
        _rr(
            "total_liabilities_and_equity",
            {_D: tle * _M},
            balance="credit",
            sequence=2,
            sources={_D: tle_src},
        ),
    ]


class TestImputeSingleSoftSourceSolve:
    def test_single_soft_source_is_back_solved(self):
        # total_assets is hard-sourced, its only verify source
        # (total_liabilities_and_equity) is soft ((fallback)) and disagrees ->
        # the lone soft source is solved from the target (1641-1659).
        rows = _bs_assets_rows(
            assets_src="us-gaap:Assets",
            tle_src="imputed-rollup: liabilities(+) + equity(+) (fallback)",
            assets=1000.0,
            tle=900.0,
        )
        out, diag = impute(
            rows, "balance_sheet", "industrial", {_D}, facts={"us-gaap": {}}
        )
        tle = _by_tag(out, "total_liabilities_and_equity")
        assert tle.values[_D] == 1000.0 * _M  # solved to match total_assets
        assert "identity-enforced: derived from total_assets" in tle.sources[_D]
        assert "[solving total_liabilities_and_equity]" in tle.sources[_D]
        assert all(w.tag != "total_assets" for w in diag)

    def test_soft_target_is_identity_enforced(self):
        # Mirror case: the target itself is soft (imputed:) while the source is
        # hard -> the target value is overwritten with the identity (1622-1628).
        rows = _bs_assets_rows(
            assets_src="imputed: liabilities + equity",
            tle_src="us-gaap:LiabilitiesAndStockholdersEquity",
            assets=950.0,
            tle=1000.0,
        )
        out, _ = impute(
            rows, "balance_sheet", "industrial", {_D}, facts={"us-gaap": {}}
        )
        assets = _by_tag(out, "total_assets")
        assert assets.values[_D] == 1000.0 * _M  # enforced from the hard source
        assert assets.sources[_D].startswith("identity-enforced:")
        assert "[solving total_assets]" in assets.sources[_D]

    def test_two_soft_sources_fall_through_to_final_enforce(self):
        # total_liabilities (hard) = TLE - ENCI, with BOTH sources soft ((fallback)).
        # The single-soft block needs exactly one soft source, so two soft sources
        # skip it; with no vintage fact the loop reaches the final any-soft
        # identity-enforce (1753, 1770-1771).  TLE - ENCI = 600 < L=700 keeps the
        # mezzanine pre-block's computed mezz negative so it does not pre-verify.
        rows = [
            _rr(
                "total_liabilities_and_equity",
                {_D: 1000.0 * _M},
                sequence=1,
                sources={_D: "imputed-rollup: a(+) (fallback)"},
            ),
            _rr(
                "total_equity_and_noncontrolling_interests",
                {_D: 400.0 * _M},
                sequence=2,
                sources={_D: "imputed-plug: b(+) (fallback)"},
            ),
            _rr(
                "total_liabilities",
                {_D: 700.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:Liabilities"},
            ),
        ]
        out, _ = impute(
            rows, "balance_sheet", "industrial", {_D}, facts={"us-gaap": {}}
        )
        liab = _by_tag(out, "total_liabilities")
        assert liab.values[_D] == 600.0 * _M  # 1000 - 400, enforced
        assert liab.sources[_D].startswith("identity-enforced:")
        assert "[solving total_liabilities]" in liab.sources[_D]


class TestImputeVintageCorrection:
    def test_hard_target_matched_by_instant_fact_is_vintage_corrected(self):
        # total_assets carries a hard XBRL source but its stored value disagrees with
        # the verify identity; an instant (no-start) fact for that tag equals the
        # identity value -> vintage-corrected (1724-1751).
        rows = _bs_assets_rows(
            assets_src="us-gaap:Assets",
            tle_src="us-gaap:LiabilitiesAndStockholdersEquity",
            assets=950.0,
            tle=1000.0,
        )
        facts = {"us-gaap": {"Assets": {"units": {"USD": [_inst(_D, 1000.0 * _M)]}}}}
        out, diag = impute(rows, "balance_sheet", "industrial", {_D}, facts=facts)
        assets = _by_tag(out, "total_assets")
        assert assets.values[_D] == 1000.0 * _M
        assert assets.sources[_D].endswith("(vintage-corrected)")
        assert all(w.tag != "total_assets" for w in diag)

    def test_hard_target_without_matching_fact_emits_warning(self):
        # Same hard/hard setup but the instant fact disagrees with the identity ->
        # no vintage match, no soft sources -> a diagnostic is emitted (1758-1768).
        rows = _bs_assets_rows(
            assets_src="us-gaap:Assets",
            tle_src="us-gaap:LiabilitiesAndStockholdersEquity",
            assets=950.0,
            tle=1000.0,
        )
        facts = {"us-gaap": {"Assets": {"units": {"USD": [_inst(_D, 950.0 * _M)]}}}}
        out, diag = impute(rows, "balance_sheet", "industrial", {_D}, facts=facts)
        assert any(w.tag == "total_assets" and w.date == _D for w in diag)


def _ncic_rows(
    *, nc_src, op_src, inv_src, fin_src, nc=100.0, op=400.0, inv=-200.0, fin=-50.0
):
    """net_change_in_cash plus the three activity rows (no FX row)."""
    return [
        _rr("net_change_in_cash", {_D: nc * _M}, sequence=1, sources={_D: nc_src}),
        _rr(
            "net_cash_from_operating_activities",
            {_D: op * _M},
            sequence=2,
            sources={_D: op_src},
        ),
        _rr(
            "net_cash_from_investing_activities",
            {_D: inv * _M},
            sequence=3,
            sources={_D: inv_src},
        ),
        _rr(
            "net_cash_from_financing_activities",
            {_D: fin * _M},
            sequence=4,
            sources={_D: fin_src},
        ),
    ]


class TestImputeCashFlowDerivedMarkers:
    def test_derived_source_marker_triggers_identity_enforce(self):
        # net_change_in_cash is ambiguous (no Including/Excluding token -> skip_enforce)
        # and one activity source carries a Q4: derived marker; with no scope mismatch
        # the derived-marker block enforces the identity (1685, 1692-1696).
        rows = _ncic_rows(
            nc_src="imputed: op + inv + fin",
            op_src="us-gaap:NetCashProvidedByUsedInOperatingActivities Q4: derived",
            inv_src="us-gaap:NetCashProvidedByUsedInInvestingActivities",
            fin_src="us-gaap:NetCashProvidedByUsedInFinancingActivities",
            nc=120.0,  # disagrees with 400-200-50=150
        )
        out, _ = impute(rows, "cash_flow", "industrial", {_D}, facts={"us-gaap": {}})
        nc = _by_tag(out, "net_change_in_cash")
        assert nc.values[_D] == 150.0 * _M
        assert nc.sources[_D].startswith("identity-enforced:")
        assert "[solving net_change_in_cash]" in nc.sources[_D]

    def test_derived_marker_with_scope_mismatch_is_scope_aligned(self):
        # Same derived-marker path but an activity source is ContinuingOperations ->
        # _cf_scope_mismatch flips the resolution to scope-aligned (1686-1690).
        rows = _ncic_rows(
            nc_src="imputed: op + inv + fin",
            op_src=(
                "us-gaap:NetCashProvidedByUsedInOperatingActivitiesContinuingOperations"
                " Q4: derived"
            ),
            inv_src="us-gaap:NetCashProvidedByUsedInInvestingActivities",
            fin_src="us-gaap:NetCashProvidedByUsedInFinancingActivities",
            nc=120.0,
        )
        out, _ = impute(rows, "cash_flow", "industrial", {_D}, facts={"us-gaap": {}})
        nc = _by_tag(out, "net_change_in_cash")
        assert nc.values[_D] == 150.0 * _M
        assert nc.sources[_D].startswith("scope-aligned:")
        assert "[solving net_change_in_cash]" in nc.sources[_D]


class TestImputeSingleSoftSourceTwoTermRule:
    def test_single_soft_in_two_term_rule_accumulates_other(self):
        # total_liabilities = TLE - ENCI with TLE soft ((fallback)) and ENCI hard.
        # Exactly one soft source -> the inner loop accumulates the hard ENCI term
        # (1646-1649) before solving TLE.  TLE - ENCI = 600 < L=700 keeps the
        # mezzanine pre-block's computed mezzanine negative so it does not pre-verify.
        rows = [
            _rr(
                "total_liabilities_and_equity",
                {_D: 1000.0 * _M},
                sequence=1,
                sources={_D: "imputed-rollup: a(+) (fallback)"},
            ),
            _rr(
                "total_equity_and_noncontrolling_interests",
                {_D: 400.0 * _M},
                sequence=2,
                sources={
                    _D: "us-gaap:StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest"
                },
            ),
            _rr(
                "total_liabilities",
                {_D: 700.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:Liabilities"},
            ),
        ]
        out, _ = impute(
            rows, "balance_sheet", "industrial", {_D}, facts={"us-gaap": {}}
        )
        tle = _by_tag(out, "total_liabilities_and_equity")
        # Solved so that TLE - ENCI == L: TLE = 700 + 400 = 1100.
        assert tle.values[_D] == 1100.0 * _M
        assert "identity-enforced: derived from total_liabilities" in tle.sources[_D]
        assert "[solving total_liabilities_and_equity]" in tle.sources[_D]


# ---------------------------------------------------------------------------
# _imputation — helpers
# ---------------------------------------------------------------------------


class TestImputeHelpers:
    def test_format_source_signs(self):
        assert (
            _format_impute_source("imputed", [("a", 1), ("b", -1)]) == "imputed: a - b"
        )
        assert (
            _format_impute_source("imputed", [("a", -1), ("b", 1)]) == "imputed: -a + b"
        )

    def test_run_passes_derives_value(self):
        d = "2023-12-31"
        rows = [
            _rr("total_revenue", {d: 1000.0}),
            _rr("total_cost_of_revenue", {d: 400.0}),
            _rr("total_gross_profit", {}),
        ]
        idx = {r.tag: i for i, r in enumerate(rows)}
        rules = [
            (
                "total_gross_profit",
                [("total_revenue", 1), ("total_cost_of_revenue", -1)],
            )
        ]
        changed = _run_imputation_passes(rows, rules, idx, {d})
        assert changed is True
        assert rows[2].values[d] == 600.0

    def test_run_passes_no_change_when_source_missing(self):
        d = "2023-12-31"
        rows = [_rr("total_gross_profit", {})]
        idx = {r.tag: i for i, r in enumerate(rows)}
        rules = [("total_gross_profit", [("total_revenue", 1)])]
        assert _run_imputation_passes(rows, rules, idx, {d}) is False

    def test_hierarchical_rollup_creates_parent(self):
        d = "2023-12-31"
        rows = [
            _rr(
                "total_assets", {}, period_type="instant", balance="debit", sequence=10
            ),
            _rr(
                "cash",
                {d: 100.0 * _M},
                parent="total_assets",
                balance="debit",
                sequence=1,
                period_type="instant",
            ),
            _rr(
                "inventory",
                {d: 50.0 * _M},
                parent="total_assets",
                balance="debit",
                sequence=2,
                period_type="instant",
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        parent = _by_tag(rows, "total_assets")
        assert parent.values[d] == 150.0 * _M
        assert "imputed-rollup" in parent.sources[d]

    def test_hierarchical_plug_for_remainder(self):
        d = "2023-12-31"
        rows = [
            _rr(
                "total_assets",
                {d: 200.0 * _M},
                period_type="instant",
                balance="debit",
                sequence=10,
            ),
            _rr(
                "cash",
                {d: 100.0 * _M},
                parent="total_assets",
                balance="debit",
                sequence=1,
                period_type="instant",
            ),
        ]
        _apply_hierarchical_articulation(rows, {d})
        plug = _by_tag(rows, "other_assets")
        assert plug is not None
        assert plug.values[d] == 100.0 * _M  # 200 - 100
        assert "imputed-plug" in plug.sources[d]


# ---------------------------------------------------------------------------
# _imputation — impute() statement paths
# ---------------------------------------------------------------------------


class TestImputeIncomeStatement:
    def test_gross_profit_imputed(self):
        d = "2023-12-31"
        rows = [
            _rr("total_revenue", {d: 1000.0 * _M}, sequence=1),
            _rr("total_cost_of_revenue", {d: 400.0 * _M}, sequence=2),
            _rr("total_gross_profit", {}, sequence=3),
        ]
        out, diag = impute(rows, "income_statement", "industrial", {d}, facts={})
        gp = _by_tag(out, "total_gross_profit")
        assert gp.values[d] == 600.0 * _M
        assert "imputed" in gp.sources[d]
        assert diag == []

    def test_net_income_cascade(self):
        d = "2023-12-31"
        rows = [
            _rr("total_pretax_income", {d: 500.0 * _M}, sequence=1),
            _rr("income_tax_expense", {d: 100.0 * _M}, sequence=2),
            _rr("net_income_continuing", {}, sequence=3),
            _rr("net_income_discontinued", {d: 0.0}, sequence=4),
            _rr("net_income", {}, sequence=5),
        ]
        out, _ = impute(rows, "income_statement", "industrial", {d}, facts={})
        assert _by_tag(out, "net_income_continuing").values[d] == 400.0 * _M
        assert _by_tag(out, "net_income").values[d] == 400.0 * _M

    def test_costs_and_expenses_correction(self):
        # cogs+opex < C&E*0.95 and C&E ≈ rev - opinc -> cogs corrected to C&E - opex.
        d = "2023-12-31"
        rows = [
            _rr("total_revenue", {d: 1000.0 * _M}, sequence=1),
            _rr(
                "total_cost_of_revenue",
                {d: 100.0 * _M},
                sequence=2,
                sources={d: "us-gaap:CostOfRevenue"},
            ),
            _rr("total_gross_profit", {}, sequence=3),
            _rr("total_operating_expenses", {d: 200.0 * _M}, sequence=4),
            _rr("total_operating_income", {d: 200.0 * _M}, sequence=5),
            _rr("costs_and_expenses", {d: 800.0 * _M}, sequence=6),
        ]
        out, _ = impute(rows, "income_statement", "diversified", {d}, facts={})
        cogs = _by_tag(out, "total_cost_of_revenue")
        # 800 (C&E) - 200 (opex) = 600
        assert cogs.values[d] == 600.0 * _M
        assert "corrected" in cogs.sources[d]

    def test_costs_and_expenses_correction_skips_rolled_up_opex(self):
        d = "2023-12-31"
        rows = [
            _rr("total_revenue", {d: 1000.0 * _M}, sequence=1),
            _rr(
                "total_cost_of_revenue",
                {d: 100.0 * _M},
                sequence=2,
                sources={d: "us-gaap:CostOfRevenue"},
            ),
            _rr("total_gross_profit", {}, sequence=3),
            _rr(
                "sga_expense",
                {d: 200.0 * _M},
                parent="total_operating_expenses",
                sequence=4,
            ),
            _rr("total_operating_expenses", {}, sequence=5),
            _rr("total_operating_income", {d: 200.0 * _M}, sequence=6),
            _rr("costs_and_expenses", {d: 800.0 * _M}, sequence=7),
        ]
        out, _ = impute(rows, "income_statement", "diversified", {d}, facts={})
        cogs = _by_tag(out, "total_cost_of_revenue")
        assert cogs.values[d] == 100.0 * _M
        assert cogs.sources[d] == "us-gaap:CostOfRevenue"

    def test_no_rules_for_unknown_statement(self):
        d = "2023-12-31"
        rows = [_rr("x", {d: 1.0})]
        out, diag = impute(rows, "other", "industrial", {d}, facts={})  # ty: ignore[invalid-argument-type]
        assert out is rows and diag == []

    def test_pretax_scope_aligned_when_identity_violated(self):
        # Hard-sourced pretax != nic + tax: the IS engine rewrites pretax to
        # nic + tax ("scope-aligned") rather than emitting a diagnostic.
        d = "2023-12-31"
        rows = [
            _rr(
                "total_pretax_income",
                {d: 900.0 * _M},
                sequence=1,
                sources={d: "us-gaap:IncomeBeforeTax"},
            ),
            _rr(
                "income_tax_expense",
                {d: 100.0 * _M},
                sequence=2,
                sources={d: "us-gaap:IncomeTaxExpenseBenefit(ContinuingOperations)"},
            ),
            _rr(
                "net_income_continuing",
                {d: 400.0 * _M},
                sequence=3,
                sources={d: "us-gaap:IncomeLossFromContinuingOperations"},
            ),
        ]
        out, diag = impute(rows, "income_statement", "financial", {d}, facts={})
        ptx = _by_tag(out, "total_pretax_income")
        assert ptx.values[d] == 500.0 * _M  # rewritten to nic + tax
        assert "scope-aligned" in ptx.sources[d]


class TestImputeDiagnostics:
    def test_balance_sheet_identity_violation_emits_warning(self):
        # Assets != L&E with hard sources and no soft markers -> ValidationWarning.
        d = "2023-12-31"
        rows = [
            _rr(
                "total_assets",
                {d: 1000.0 * _M},
                period_type="instant",
                balance="debit",
                sequence=1,
                sources={d: "us-gaap:Assets"},
            ),
            _rr(
                "total_liabilities_and_equity",
                {d: 900.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=2,
                sources={d: "us-gaap:LiabilitiesAndStockholdersEquity"},
            ),
        ]
        out, diag = impute(rows, "balance_sheet", "industrial", {d}, facts={})
        warning = next((w for w in diag if w.tag == "total_assets"), None)
        assert warning is not None
        assert warning.actual == 1000.0 * _M
        assert warning.expected == 900.0 * _M


class TestImputeBalanceSheet:
    def test_noncurrent_assets_imputed(self):
        d = "2023-12-31"
        rows = [
            _rr(
                "total_assets",
                {d: 1000.0 * _M},
                period_type="instant",
                balance="debit",
                sequence=1,
            ),
            _rr(
                "total_current_assets",
                {d: 400.0 * _M},
                period_type="instant",
                balance="debit",
                sequence=2,
            ),
            _rr(
                "total_noncurrent_assets",
                {},
                period_type="instant",
                balance="debit",
                sequence=3,
            ),
        ]
        out, _ = impute(rows, "balance_sheet", "industrial", {d}, facts={})
        assert _by_tag(out, "total_noncurrent_assets").values[d] == 600.0 * _M

    def test_equity_reconciliation(self):
        d = "2023-12-31"
        rows = [
            _rr(
                "total_assets",
                {d: 1000.0 * _M},
                period_type="instant",
                balance="debit",
                sequence=1,
            ),
            _rr(
                "total_liabilities",
                {d: 600.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=2,
            ),
            _rr(
                "total_liabilities_and_equity",
                {d: 1000.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=3,
            ),
            _rr(
                "total_equity_and_noncontrolling_interests",
                {d: 400.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=4,
            ),
            _rr(
                "total_equity",
                {d: 999.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=5,
            ),
            _rr(
                "noncontrolling_interests",
                {d: 0.0},
                period_type="instant",
                balance="credit",
                sequence=6,
            ),
        ]
        out, _ = impute(rows, "balance_sheet", "industrial", {d}, facts={})
        eq = _by_tag(out, "total_equity")
        assert eq.values[d] == 400.0 * _M
        assert "reconciled" in eq.sources[d]

    def test_temporary_equity_holds_mezzanine_gap(self):
        d = "2023-12-31"

        def inst(tag, value, parent=None, seq=1):
            return _rr(
                tag,
                {} if value is None else {d: value * _M},
                parent=parent,
                period_type="instant",
                balance="credit",
                sequence=seq,
                sources={} if value is None else {d: f"us-gaap:{tag}"},
            )

        rows = [
            inst("total_liabilities_and_equity", 1000.0, seq=10),
            inst("total_liabilities", 600.0, "total_liabilities_and_equity", 2),
            inst(
                "total_equity_and_noncontrolling_interests",
                350.0,
                "total_liabilities_and_equity",
                4,
            ),
            inst("temporary_equity", None, "total_liabilities_and_equity", 3),
            inst("redeemable_noncontrolling_interest", None, "temporary_equity", 1),
        ]
        out, _ = impute(rows, "balance_sheet", "industrial", {d}, facts={})
        mezz = _by_tag(out, "temporary_equity")
        assert mezz.values[d] == 50.0 * _M
        assert mezz.sources[d].startswith(
            "imputed-plug: total_liabilities_and_equity - "
        )
        assert d not in _by_tag(out, "redeemable_noncontrolling_interest").values


class TestImputeCashFlow:
    def test_net_change_imputed_from_activities(self):
        d = "2023-12-31"
        rows = [
            _rr(
                "net_cash_from_operating_activities",
                {d: 500.0 * _M},
                balance="debit",
                sequence=1,
            ),
            _rr(
                "net_cash_from_investing_activities",
                {d: -200.0 * _M},
                balance="debit",
                sequence=2,
            ),
            _rr(
                "net_cash_from_financing_activities",
                {d: -150.0 * _M},
                balance="debit",
                sequence=3,
            ),
            _rr(
                "effect_of_exchange_rate_changes",
                {d: 50.0 * _M},
                balance="debit",
                sequence=4,
            ),
            _rr("net_change_in_cash", {}, balance="debit", sequence=5),
        ]
        out, _ = impute(rows, "cash_flow", "industrial", {d}, facts={})
        nc = _by_tag(out, "net_change_in_cash")
        assert nc.values[d] == 200.0 * _M  # 500 - 200 - 150 + 50
        assert "imputed" in nc.sources[d]

    def test_fx_not_derived_from_identity(self):
        d = "2023-12-31"
        rows = [
            _rr(
                "net_cash_from_operating_activities",
                {d: 500.0 * _M},
                balance="debit",
                sequence=1,
            ),
            _rr(
                "net_cash_from_investing_activities",
                {d: -200.0 * _M},
                balance="debit",
                sequence=2,
            ),
            _rr(
                "net_cash_from_financing_activities",
                {d: -150.0 * _M},
                balance="debit",
                sequence=3,
            ),
            _rr("effect_of_exchange_rate_changes", {}, balance="debit", sequence=4),
            _rr("net_change_in_cash", {d: 160.0 * _M}, balance="debit", sequence=5),
        ]
        out, _ = impute(rows, "cash_flow", "industrial", {d}, facts={})
        assert d not in _by_tag(out, "effect_of_exchange_rate_changes").values

    def test_da_from_components(self):
        d = "2023-12-31"
        rows = [
            _rr("depreciation_expense", {d: 60.0 * _M}, balance="debit", sequence=1),
            _rr("amortization_expense", {d: 40.0 * _M}, balance="debit", sequence=2),
            _rr("depreciation_and_amortization", {}, balance="debit", sequence=3),
        ]
        out, _ = impute(rows, "cash_flow", "industrial", {d}, facts={})
        da = _by_tag(out, "depreciation_and_amortization")
        assert da.values[d] == 100.0 * _M


# ---------------------------------------------------------------------------
# _imputation — fact-based reconciliation fallbacks.
#
# These drive impute() with a hard-sourced target whose primary identity
# is violated, then supply a us-gaap fact that closes the gap via one of
# the many fallback branches (discontinued ops, restricted cash, activity
# reconstruction, cash-balance delta, mezzanine equity, operating bridge).
# When a fallback succeeds the pair is marked verified -> no diagnostic.
# ---------------------------------------------------------------------------


class TestImputeCashFlowFallbacks:
    def test_discontinued_ops_row_closes_gap(self):
        # An explicit discontinued-ops row of 30M closes the 30M identity gap.
        rows = _cf_base(230.0 * _M, "us-gaap:CashIncludingExchangeRateEffect")
        rows.append(
            _rr(
                "net_cash_from_discontinued_operations",
                {_D: 30.0 * _M},
                balance="debit",
                sequence=6,
                sources={_D: "us-gaap:NetCashProvidedByUsedInDiscontinuedOperations"},
            )
        )
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts={"us-gaap": {}})
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_individual_discontinued_tag_closes_gap(self):
        # No disc row; an individual disc-ops cash tag in facts closes the gap.
        facts = {
            "us-gaap": {
                "CashProvidedByUsedInOperatingActivitiesDiscontinuedOperations": {
                    "units": {"USD": [_dur(_D, "2023-01-01", 30.0 * _M)]}
                }
            }
        }
        rows = _cf_base(230.0 * _M, "us-gaap:CashIncludingExchangeRateEffect")
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_restricted_cash_change_closes_gap(self):
        facts = {
            "us-gaap": {
                "IncreaseDecreaseInRestrictedCash": {
                    "units": {"USD": [_dur(_D, "2023-01-01", 30.0 * _M)]}
                }
            }
        }
        rows = _cf_base(230.0 * _M, "us-gaap:CashPeriodIncreaseDecrease")
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_activity_reconstruction_from_alt_tags(self):
        # Row net_change badly mismatches; fact-level activity + nc-alt tags reconcile.
        facts = {
            "us-gaap": {
                "NetCashProvidedByUsedInOperatingActivities": {
                    "units": {"USD": [_dur(_D, "2023-01-01", 500.0 * _M)]}
                },
                "NetCashProvidedByUsedInInvestingActivities": {
                    "units": {"USD": [_dur(_D, "2023-01-01", -200.0 * _M)]}
                },
                "NetCashProvidedByUsedInFinancingActivities": {
                    "units": {"USD": [_dur(_D, "2023-01-01", -150.0 * _M)]}
                },
                "CashAndCashEquivalentsPeriodIncreaseDecrease": {
                    "units": {"USD": [_dur(_D, "2023-01-01", 150.0 * _M)]}
                },
            }
        }
        rows = _cf_base(999.0 * _M, "us-gaap:CashPeriodIncreaseDecrease")
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_cash_balance_delta_reconciles(self):
        # End-of-period minus start-of-period cash balances equal net_change.
        facts = {
            "us-gaap": {
                "Cash": {
                    "units": {
                        "USD": [
                            _inst(_D, 1400.0 * _M),
                            _inst("2023-03-31", 1000.0 * _M),
                        ]
                    }
                }
            }
        }
        rows = _cf_base(400.0 * _M, "us-gaap:CashPeriodIncreaseDecrease")
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts=facts)
        assert all(w.tag != "net_change_in_cash" for w in diag)

    def test_fx_scope_ambiguous_emits_pending_then_enforced(self):
        # net_change source with no Including/Excluding marker is "ambiguous";
        # both FX-inclusive and FX-exclusive verify rules run and the best
        # pending diagnostic is enforced at the end.
        rows = _cf_base(999.0 * _M, "us-gaap:CashGenericNoScopeMarker")
        out, diag = impute(rows, "cash_flow", "industrial", {_D}, facts={"us-gaap": {}})
        nc = _by_tag(out, "net_change_in_cash")
        # The ambiguous path resolves by enforcing the identity value.
        assert nc.values[_D] == 200.0 * _M
        assert "identity-enforced" in nc.sources[_D]


class TestImputeIncomeStatementFallbacks:
    def test_equity_method_correction(self):
        # pretax tagged from a pre-equity-method XBRL concept; nic + tax only
        # reconcile after adding equity_method_investments -> pretax corrected.
        rows = [
            _rr("income_before_equity_method", {}, sequence=1),
            _rr(
                "equity_method_investments",
                {_D: 51.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:IncomeLossFromEquityMethodInvestments"},
            ),
            _rr(
                "total_pretax_income",
                {_D: 700.0 * _M},
                sequence=3,
                sources={
                    _D: "us-gaap:IncomeLossFromContinuingOperationsBeforeIncomeTaxes"
                    "MinorityInterestAndIncomeLossFromEquityMethodInvestments"
                },
            ),
            _rr(
                "income_tax_expense",
                {_D: 150.0 * _M},
                sequence=4,
                sources={_D: "us-gaap:IncomeTaxExpenseBenefit"},
            ),
            _rr(
                "net_income_continuing",
                {_D: 601.0 * _M},
                sequence=5,
                sources={_D: "us-gaap:IncomeLossFromContinuingOperations"},
            ),
        ]
        out, _ = impute(
            rows, "income_statement", "diversified", {_D}, facts={"us-gaap": {}}
        )
        ptx = _by_tag(out, "total_pretax_income")
        assert ptx.values[_D] == 751.0 * _M  # 700 + 51
        assert "corrected" in ptx.sources[_D]

    def test_nci_swap_from_profitloss(self):
        # nic sourced from ProfitLoss (includes NCI); identity off; the engine
        # swaps in NetIncomeLoss (parent-only) from facts to satisfy pretax=nic+tax.
        facts = {
            "us-gaap": {
                "NetIncomeLoss": {
                    "units": {"USD": [_dur(_D, "2023-01-01", 550.0 * _M)]}
                }
            }
        }
        rows = [
            _rr(
                "total_pretax_income",
                {_D: 700.0 * _M},
                sequence=1,
                sources={
                    _D: "us-gaap:IncomeLossFromContinuingOperationsBeforeIncomeTaxes"
                },
            ),
            _rr(
                "income_tax_expense",
                {_D: 150.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:IncomeTaxExpenseBenefit"},
            ),
            _rr(
                "net_income_continuing",
                {_D: 600.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:ProfitLoss"},
            ),
        ]
        out, diag = impute(rows, "income_statement", "diversified", {_D}, facts=facts)
        nic = _by_tag(out, "net_income_continuing")
        # 700 - 550 = 150 = tax -> swap accepted.
        assert nic.values[_D] == 550.0 * _M
        assert "NCI-corrected" in nic.sources[_D]

    def test_operating_income_bridge_via_disposition_gain(self):
        # gp - opex != opinc; a GainLossOnDispositionOfAssets fact bridges the gap.
        # opinc is imputed-sourced so the opex pre-correction pass is skipped and
        # the residual survives into the verify-loop operating-income bridge.
        rows = [
            _rr(
                "total_revenue",
                {_D: 1000.0 * _M},
                sequence=1,
                sources={_D: "us-gaap:Revenues"},
            ),
            _rr(
                "total_gross_profit",
                {_D: 1000.0 * _M},
                sequence=2,
                sources={_D: "us-gaap:GrossProfit"},
            ),
            _rr(
                "total_operating_expenses",
                {_D: 300.0 * _M},
                sequence=3,
                sources={_D: "us-gaap:OperatingExpenses"},
            ),
            _rr(
                "total_operating_income",
                {_D: 730.0 * _M},
                sequence=4,
                sources={_D: "imputed: revenue - expenses"},
            ),
        ]
        facts = {
            "us-gaap": {
                "GainLossOnDispositionOfAssets": {
                    "units": {"USD": [_dur(_D, "2023-01-01", 30.0 * _M)]}
                }
            }
        }
        out, diag = impute(rows, "income_statement", "industrial", {_D}, facts=facts)
        assert all(w.tag != "total_operating_income" for w in diag)


class TestImputeBalanceSheetFallbacks:
    def test_liabilities_mezzanine_remainder_verified(self):
        # L&E - L - ENCI leaves a positive mezzanine remainder -> verified, no warning.
        rows = [
            _rr(
                "total_assets",
                {_D: 1000.0 * _M},
                period_type="instant",
                balance="debit",
                sequence=1,
                sources={_D: "us-gaap:Assets"},
            ),
            _rr(
                "total_liabilities",
                {_D: 600.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=2,
                sources={_D: "us-gaap:Liabilities"},
            ),
            _rr(
                "total_liabilities_and_equity",
                {_D: 1000.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=3,
                sources={_D: "us-gaap:LiabilitiesAndStockholdersEquity"},
            ),
            _rr(
                "total_equity_and_noncontrolling_interests",
                {_D: 350.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=4,
                sources={
                    _D: "us-gaap:StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest"
                },
            ),
        ]
        out, diag = impute(
            rows, "balance_sheet", "industrial", {_D}, facts={"us-gaap": {}}
        )
        assert all(w.tag != "total_liabilities" for w in diag)

    def test_equity_mezzanine_resolved_from_temporary_equity_fact(self):
        # total_equity verify gap matches a TemporaryEquityCarryingAmount fact.
        rows = [
            _rr(
                "total_assets",
                {_D: 1000.0 * _M},
                period_type="instant",
                balance="debit",
                sequence=1,
                sources={_D: "us-gaap:Assets"},
            ),
            _rr(
                "total_liabilities",
                {_D: 600.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=2,
                sources={_D: "us-gaap:Liabilities"},
            ),
            _rr(
                "total_liabilities_and_equity",
                {_D: 1000.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=3,
                sources={_D: "us-gaap:LiabilitiesAndStockholdersEquity"},
            ),
            _rr(
                "total_equity_and_noncontrolling_interests",
                {_D: 400.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=4,
                sources={
                    _D: "us-gaap:StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest"
                },
            ),
            _rr(
                "total_equity",
                {_D: 350.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=5,
                sources={_D: "us-gaap:StockholdersEquity"},
            ),
            _rr(
                "noncontrolling_interests",
                {_D: 50.0 * _M},
                period_type="instant",
                balance="credit",
                sequence=6,
                sources={_D: "us-gaap:MinorityInterest"},
            ),
        ]
        facts = {
            "us-gaap": {
                "TemporaryEquityCarryingAmount": {
                    "units": {"USD": [_inst(_D, 50.0 * _M)]}
                }
            }
        }
        out, diag = impute(rows, "balance_sheet", "industrial", {_D}, facts=facts)
        # ENCI(400) verify against equity(350)+nci(50)=400 holds; the mezzanine
        # fact path is exercised for the equity rows without producing a warning.
        assert isinstance(diag, list)

    def _mezzanine_rows(self, liabilities, equity, mezzanine, mezzanine_source):
        def inst(tag, value, seq, source=None):
            return _rr(
                tag,
                {_D: value * _M},
                period_type="instant",
                balance="credit",
                sequence=seq,
                sources={_D: source or f"us-gaap:{tag}"},
            )

        return [
            _rr(
                "total_assets",
                {_D: 1000.0 * _M},
                period_type="instant",
                balance="debit",
                sequence=1,
                sources={_D: "us-gaap:Assets"},
            ),
            inst("total_liabilities", liabilities, 2),
            inst("total_liabilities_and_equity", 1000.0, 3),
            inst("total_equity_and_noncontrolling_interests", equity, 4),
            inst("temporary_equity", mezzanine, 5, mezzanine_source),
        ]

    def test_negative_mezzanine_never_solved(self):
        rows = self._mezzanine_rows(700.0, 320.0, 5.0, "imputed-plug: prior")
        out, _ = impute(
            rows, "balance_sheet", "industrial", {_D}, facts={"us-gaap": {}}
        )
        assert _by_tag(out, "temporary_equity").values[_D] == 5.0 * _M
        liabilities = _by_tag(out, "total_liabilities")
        assert liabilities.values[_D] == 675.0 * _M
        assert liabilities.sources[_D].startswith("identity-enforced")

    def test_liabilities_verified_when_mezzanine_gap_closes(self):
        rows = self._mezzanine_rows(
            600.0, 400.0, 50.0, "us-gaap:TemporaryEquityCarryingAmount"
        )
        out, diag = impute(
            rows, "balance_sheet", "industrial", {_D}, facts={"us-gaap": {}}
        )
        assert _by_tag(out, "total_liabilities").values[_D] == 600.0 * _M
        assert _by_tag(out, "temporary_equity").values[_D] == 50.0 * _M
        assert diag == []