"cash_flows_quarterly"))

    secondary_source = None
    secondary_by_period: dict = {}
    secondary_balance: dict = {}
    secondary_cash: dict = {}
    if secondary_adapter is not None:
        if secondary_adapter.tier == "DEMO" or is_demo:
            # Mirrors _assert_no_demo_live_mix's own rule at the single-call level: a demo source
            # can never be one half of a dual-source resolution, since demo data is synthetic and
            # was never meant to be cross-checked against (or corrupt) real provider data.
            raise RuntimeError("Refusing dual-source ingestion: DEMO cannot be paired with a second provider.")
        secondary_source = _get_or_create_source(db, secondary_adapter.name, secondary_adapter.tier, is_demo, f"ingest:{ticker}")
        # AUDIT (Part A6): period alignment across two providers is keyed on (period_end,
        # period_type) only -- NOT filing_date, which two providers can legitimately report a day
        # or more apart for the "same" filing. This is a real, documented limitation: a provider
        # that reports a fiscal period boundary a few days off from the other (a genuine, if rare,
        # real-world occurrence) will not be matched and that period is silently treated as
        # primary-only for this run, not flagged as a mismatch. See docs/AUDIT_PROVIDER_CONFLICT_A6.md.
        try:
            secondary_by_period = {
                (p.period_end, p.period_type): p for p in secondary_adapter.get_income_statements(ticker)
            }
            # Part A9: the secondary provider's balance-sheet and cash-flow responses are fetched
            # too, so cross-source conflict resolution covers all four statement groups rather
            # than only the income statement. _fetch_statements_by_period returns {} for an
            # adapter that has not implemented a statement, so this adds no new failure mode.
            secondary_balance = _fetch_statements_by_period(
                secondary_adapter, secondary_adapter.get_balance_sheets, ticker, "balance_sheets")
            secondary_cash = _fetch_statements_by_period(
                secondary_adapter, secondary_adapter.get_cash_flows, ticker, "cash_flows")
        except NotImplementedError:
            # AUDIT FIX (Part A6, real bug caught before this shipped): EODHDAdapter.get_income_
            # statements/get_balance_sheets/get_cash_flows are NotImplementedError today
            # (app/adapters/eodhd.py's own module docstring -- statement-level field mapping was
            # deliberately deferred until verified against a live response). EODHD is the only
            # other real (non-DEMO) adapter besides FMP, so PROVIDER_SECONDARY="EODHD" is the one
            # realistic dual-source configuration this codebase can actually be set to today -- and
            # without this except clause, setting it would crash every single ingestion task the
            # instant this function tried to call the secondary adapter's statement method. Treated
            # as "no secondary statement data available this run", not a crash: falls back to
            # exactly the primary-only behavior secondary_adapter=None already has for every period.
            logger.warning("ingest.security.secondary_statements_not_implemented", ticker=ticker,
                            secondary_provider=secondary_adapter.name)

    for p in income_periods:
        fp = db.query(FinancialPeriod).filter_by(
            security_id=security.id, period_end=p.period_end, period_type=p.period_type, filing_date=p.filing_date,
        ).one_or_none()
        if fp is None:
            fp = FinancialPeriod(
                security_id=security.id, period_end=p.period_end, period_type=p.period_type,
                filing_date=p.filing_date, currency=p.currency, source_id=source.id,
            )
            db.add(fp)
            db.flush()

        key = (p.period_end, p.period_type)
        li = merge_statement_line_items(
            p.line_items,
            balance_by_period.get(key),
            cash_by_period.get(key),
        )
        secondary_p = secondary_by_period.get(key)
        li_b = (
            merge_statement_line_items(
                secondary_p.line_items, secondary_balance.get(key), secondary_cash.get(key)
            )
            if secondary_p is not None
            else None
        )
        secondary_tier = secondary_adapter.tier if (secondary_adapter is not None and li_b is not None) else None

        income = fp.income_statement or IncomeStatement(financial_period_id=fp.id)
        resolved, conflicts = resolve_line_items(
            li, li_b, _INCOME_FIELDS, adapter.tier, secondary_tier
        )
        merged_report = ValidationReport()
        resolved = validate_line_items(
            resolved,
            merged_report,
            row_key=f"{ticker}:{p.period_end}:{p.period_type}:income",
        )
        for field, value in resolved.items():
            setattr(income, field, value)
        db.add(income)
        if conflicts:
            _record_conflicts(db, "income_statements", fp.id, conflicts, source, secondary_source)

        balance = fp.balance_sheet or BalanceSheet(financial_period_id=fp.id)
        resolved, conflicts = resolve_line_items(
            li, li_b, _BALANCE_FIELDS, adapter.tier, secondary_tier
        )
        merged_report = ValidationReport()
        resolved = validate_line_items(
            resolved,
            merged_report,
            row_key=f"{ticker}:{p.period_end}:{p.period_type}:balance",
        )
        for field, value in resolved.items():
            setattr(balance, field, value)
        db.add(balance)
        if conflicts:
            _record_conflicts(db, "balance_sheets", fp.id, conflicts, source, secondary_source)

        cash_flow = fp.cash_flow or CashFlow(financial_period_id=fp.id)
        resolved, conflicts = resolve_line_items(
            li, li_b, _CASH_FLOW_FIELDS, adapter.tier, secondary_tier
        )
        merged_report = ValidationReport()
        resolved = validate_line_items(
            resolved,
            merged_report,
            row_key=f"{ticker}:{p.period_end}:{p.period_type}:cash_flow",
        )
        for field, value in resolved.items():
            setattr(cash_flow, field, value)
        db.add(cash_flow)
        if conflicts:
            _record_conflicts(db, "cash_flows", fp.id, conflicts, source, secondary_source)

        shares = fp.shares or Shares(financial_period_id=fp.id)
        resolved, conflicts = resolve_line_items(
            li, li_b, _SHARES_FIELDS, adapter.tier, secondary_tier
        )
        merged_report = ValidationReport()
        resolved = validate_line_items(
            resolved,
            merged_report,
            row_key=f"{ticker}:{p.period_end}:{p.period_type}:shares",
        )
        for field, value in resolved.items():
            setattr(shares, field, value)
        db.add(shares)
        if conflicts:
            _record_conflicts(db, "shares", fp.id, conflicts, source, secondary_source)

    # AUDIT FIX (final master pass, §21) — the fourth "engine built, never wired" defect.
    # `ProviderAdapter.get_estimates()` is implemented by FMP and by the demo adapter,
    # `ProviderEstimateRow` exists, and the `estimates` table exists with a unique constraint on
    # (security_id, period_end, metric, as_of_date). Nothing ever called it, nothing ever wrote a
    # row, and `build_snapshot_from_db()` never passed `forward_eps_estimate` — so it was None on
    # every real run and `forward_pe` and `eps_growth_forward` were PERMANENTLY NULL. `forward_pe`
    # is a member of the Valuation pillar, so every Valuation score was computed from six of its
    # seven metrics with nothing saying so.
    # Writes are idempotent against that unique constraint: an estimate already stored for the