"""Unit tests of the fundamentals engine: mapping, fiscal calendar, resolution, derivation, TTM, versioning, ratios."""
from __future__ import annotations

from datetime import date
from pathlib import Path

import pandas as pd
import pytest

from tests.conftest import PROTO_CIKS, load_gz

ROOT = Path(__file__).resolve().parents[1]


def _mods():
    from fundamentals import mapping as M
    from fundamentals import normalize as N
    from fundamentals import ratios as R
    from fundamentals import screener as SC
    return M, N, R, SC


def _fact(tag, val, start, end, accn, filed, *, fy=2024, fp="Q1", form="10-Q", unit="USD", taxonomy="us-gaap", frame=None):
    return dict(cik=1, taxonomy=taxonomy, tag=tag, unit=unit, fy=fy, fp=fp, form=form, start=start, end=end, val=val,
                accn=accn, filed=filed, frame=frame)


def _frame(rows):
    _, N, _, _ = _mods()
    df = pd.DataFrame(rows, columns=N.FACT_COLUMNS)
    for c in ("start", "end", "filed"):
        df[c] = pd.to_datetime(df[c]).dt.date
    df["fy"] = df["fy"].astype("Int64")
    return df


# ------------------------------------------------------------------------------------------- mapping
def test_mapping_is_prioritized_and_unique(app):
    M, _, _, _ = _mods()
    rows = M.mapping_rows()
    assert len(rows) > 120
    for a in M.ACCOUNTS:
        tags = [(t.taxonomy, t.tag) for t in a.tags]
        assert len(tags) == len(set(tags)), a.name
        if not a.computed:
            assert tags, a.name
        else:
            assert a.formula
    # the spec's chart of accounts, exactly
    assert [a.name for a in M.accounts_for(M.INCOME) if not a.auxiliary] == [
        "revenue", "cost_of_revenue", "gross_profit", "rnd_expense", "sga_expense", "operating_income", "interest_expense",
        "pretax_income", "income_tax", "net_income", "eps_basic", "eps_diluted", "shares_basic", "shares_diluted", "ebitda",
        "dividends_paid"]
    assert [a.name for a in M.accounts_for(M.BALANCE) if not a.auxiliary] == [
        "cash_and_equivalents", "short_term_investments", "receivables", "inventory", "total_current_assets", "ppe_net",
        "goodwill", "intangibles", "total_assets", "accounts_payable", "short_term_debt", "total_current_liabilities",
        "long_term_debt", "total_liabilities", "retained_earnings", "total_equity", "total_debt", "net_debt", "working_capital"]
    assert [a.name for a in M.accounts_for(M.CASHFLOW) if not a.auxiliary] == [
        "operating_cash_flow", "capex", "free_cash_flow", "acquisitions", "investing_cash_flow", "debt_issued", "debt_repaid",
        "buybacks", "dividends", "financing_cash_flow", "net_change_in_cash", "stock_based_compensation"]
    assert M.ACCOUNT_BY_NAME["revenue"].tags[0].tag == "Revenues"
    assert M.ACCOUNT_BY_NAME["revenue"].tags[1].tag == "RevenueFromContractWithCustomerExcludingAssessedTax"
    assert M.is_mapped("us-gaap", "Assets") and not M.is_mapped("us-gaap", "OperatingExpenses")


def test_mapping_seeded_in_db(fundamentals_data):
    from sqlalchemy import func, select

    from core.db import session
    from fundamentals import mapping as M
    from fundamentals.models import FundMapping
    with session() as s:
        n = s.scalar(select(func.count()).select_from(FundMapping).where(FundMapping.version == M.MAPPING_VERSION))
    assert n == len(M.mapping_rows())


# ------------------------------------------------------------------------------------ fiscal calendar
def test_fiscal_calendar_apple_and_microsoft(app):
    _, N, _, _ = _mods()
    apple = N.FiscalCalendar("0928", {2023: date(2023, 9, 30), 2024: date(2024, 9, 28)})
    assert apple.locate(date(2024, 3, 30)) == (2024, 2)        # Apple Q2 FY2024
    assert apple.locate(date(2023, 12, 30)) == (2024, 1)
    assert apple.locate(date(2024, 6, 29)) == (2024, 3)
    assert apple.locate(date(2024, 9, 28)) == (2024, 4)
    assert apple.locate(date(2025, 3, 29)) == (2025, 2)        # extrapolated from MMDD with ±7-day snap
    assert apple.locate(date(2025, 9, 27)) == (2025, 4)
    msft = N.FiscalCalendar("0630", {2024: date(2024, 6, 30)})
    assert msft.locate(date(2024, 12, 31)) == (2025, 2)
    assert msft.locate(date(2025, 6, 30)) == (2025, 4)
    assert msft.locate(date(2024, 9, 30)) == (2025, 1)
    # 52/53-week retailer ending late December
    shak = N.FiscalCalendar("1231", {2023: date(2023, 12, 27), 2024: date(2024, 12, 25)})
    assert shak.locate(date(2024, 3, 27)) == (2024, 1)
    assert shak.locate(date(2024, 12, 25)) == (2024, 4)


def test_calendar_quarter_and_span(app):
    _, N, _, _ = _mods()
    assert N.calendar_quarter(date(2024, 3, 30)) == "2024Q1"
    assert N.calendar_quarter(date(2023, 12, 30)) == "2023Q4"
    assert N.calendar_quarter(date(2025, 1, 3)) == "2024Q4"          # 53-week year ending in early January
    assert N.calendar_quarter(date(2024, 6, 29)) == "2024Q2"
    assert N.span_type(date(2023, 12, 31), date(2024, 3, 30)) == N.SPAN_Q
    assert N.span_type(date(2023, 10, 1), date(2024, 3, 30)) == N.SPAN_H
    assert N.span_type(date(2023, 10, 1), date(2024, 6, 29)) == N.SPAN_9M
    assert N.span_type(date(2023, 10, 1), date(2024, 9, 28)) == N.SPAN_FY
    assert N.span_type(None, date(2024, 9, 28)) is None
    assert N.span_type(date(2024, 8, 1), date(2024, 9, 28)) is None   # odd stub period


# ---------------------------------------------------------------------------------------- resolution
def _one_filing(rows, fye="1231", fy_ends=None):
    _, N, _, _ = _mods()
    facts = _frame(rows)
    filings = {"A1": N.Filing("A1", "10-Q", date(2024, 5, 1), date(2024, 3, 31))}
    cal = N.FiscalCalendar(fye, fy_ends or {2023: date(2023, 12, 31)})
    return N.resolve_filing(facts, cal, filings["A1"])


def test_priority_and_fallback(app):
    periods, _, _ = _one_filing([
        _fact("RevenueFromContractWithCustomerExcludingAssessedTax", 100, "2024-01-01", "2024-03-31", "A1", "2024-05-01"),
        _fact("SalesRevenueNet", 90, "2024-01-01", "2024-03-31", "A1", "2024-05-01"),
    ])
    inc = periods[(2024, 1, "Q")]["income"]
    assert inc.values["revenue"] == 100 and inc.coverage["revenue"]["priority"] == 2
    periods, _, _ = _one_filing([
        _fact("Revenues", 120, "2024-01-01", "2024-03-31", "A1", "2024-05-01"),
        _fact("RevenueFromContractWithCustomerExcludingAssessedTax", 100, "2024-01-01", "2024-03-31", "A1", "2024-05-01"),
    ])
    inc = periods[(2024, 1, "Q")]["income"]
    assert inc.values["revenue"] == 120 and inc.coverage["revenue"]["tag"] == "us-gaap:Revenues"


def test_component_sum_and_unit_normalisation(app):
    periods, _, _ = _one_filing([
        _fact("GeneralAndAdministrativeExpense", 30, "2024-01-01", "2024-03-31", "A1", "2024-05-01"),
        _fact("SellingAndMarketingExpense", 20, "2024-01-01", "2024-03-31", "A1", "2024-05-01"),
        _fact("EarningsPerShareDiluted", 1.5, "2024-01-01", "2024-03-31", "A1", "2024-05-01", unit="USD/shares"),
        _fact("WeightedAverageNumberOfDilutedSharesOutstanding", 1_000, "2024-01-01", "2024-03-31", "A1", "2024-05-01", unit="shares"),
        _fact("Assets", 5_000, None, "2024-03-31", "A1", "2024-05-01", unit="CAD"),
        _fact("Assets", 999, None, "2024-03-31", "A1", "2024-05-01", unit="pure"),      # wrong unit: ignored
        _fact("EarningsPerShareDiluted", 7, "2024-01-01", "2024-03-31", "A1", "2024-05-01", unit="USD"),  # wrong unit: ignored
    ])
    inc = periods[(2024, 1, "Q")]["income"]
    assert inc.values["sga_expense"] == 50 and inc.coverage["sga_expense"]["components"] is True
    assert inc.values["eps_diluted"] == 1.5 and inc.values["shares_diluted"] == 1_000
    bal = periods[(2024, 1, "I")]["balance"]
    assert bal.values["total_assets"] == 5_000 and bal.currency == "CAD" and bal.coverage["total_assets"]["currency"] == "CAD"


# -------------------------------------------------------------------------------- derivation / versions
def _company(rows, fye="1231"):
    _, N, _, _ = _mods()
    facts = _frame(rows)
    return N.normalize_company(1, "TEST", facts, {}, fye)


def test_q4_derivation_requires_three_quarters_same_year(app):
    def q(tag, val, s, e, accn, filed, fp, fy=2023, form="10-Q"):
        return _fact(tag, val, s, e, accn, filed, fy=fy, fp=fp, form=form)
    rows = [q("Revenues", 10, "2023-01-01", "2023-03-31", "Q1", "2023-05-01", "Q1"),
            q("Revenues", 20, "2023-04-01", "2023-06-30", "Q2", "2023-08-01", "Q2"),
            q("Revenues", 30, "2023-07-01", "2023-09-30", "Q3", "2023-11-01", "Q3"),
            q("Revenues", 100, "2023-01-01", "2023-12-31", "K", "2024-02-15", "FY", form="10-K"),
            q("WeightedAverageNumberOfDilutedSharesOutstanding", 100, "2023-01-01", "2023-03-31", "Q1", "2023-05-01", "Q1"),
            q("WeightedAverageNumberOfDilutedSharesOutstanding", 100, "2023-04-01", "2023-06-30", "Q2", "2023-08-01", "Q2"),
            q("WeightedAverageNumberOfDilutedSharesOutstanding", 100, "2023-07-01", "2023-09-30", "Q3", "2023-11-01", "Q3"),
            q("WeightedAverageNumberOfDilutedSharesOutstanding", 95, "2023-01-01", "2023-12-31", "K", "2024-02-15", "FY", form="10-K")]
    for r in rows:
        r["unit"] = "shares" if "Shares" in r["tag"] else "USD"
    res = _company(rows)
    q4 = [r for r in res.rows if r["statement"] == "income" and r["fiscal_quarter"] == 4]
    assert len(q4) == 1 and q4[0]["derived"] is True
    assert q4[0]["revenue"] == 40 and q4[0]["coverage"]["revenue"]["derived"] == "FY-(Q1+Q2+Q3)"
    assert q4[0]["shares_diluted"] == 4 * 95 - 300 and q4[0]["coverage"]["shares_diluted"]["approx"] is True
    assert q4[0]["form"] == "10-K" and str(q4[0]["filed_date"]) == "2024-02-15" and str(q4[0]["period_end"]) == "2023-12-31"
    # drop Q2 → Q4 cannot be derived, no invented value
    res2 = _company([r for r in rows if r["accn"] != "Q2"])
    assert not [r for r in res2.rows if r["statement"] == "income" and r["fiscal_quarter"] == 4]


def test_restatement_versioning_and_as_of(app):
    _, N, _, _ = _mods()
    rows = [_fact("Revenues", 100, "2023-01-01", "2023-03-31", "Q1", "2023-05-01", fy=2023, fp="Q1"),
            _fact("Revenues", 110, "2023-01-01", "2023-03-31", "Q1A", "2023-09-01", fy=2023, fp="Q1", form="10-Q/A"),
            _fact("Revenues", 110, "2023-01-01", "2023-03-31", "Q1N", "2024-05-01", fy=2024, fp="Q1")]   # comparative, unchanged
    res = _company(rows)
    versions = sorted([r for r in res.rows if r["statement"] == "income"], key=lambda r: r["filed_date"])
    assert [(v["revenue"], v["restated"], v["form"]) for v in versions] == [(100, False, "10-Q"), (110, True, "10-Q/A")]
    assert N.select_as_of(res.rows, date(2023, 6, 1))[0]["revenue"] == 100
    assert N.select_as_of(res.rows, date(2023, 9, 1))[0]["revenue"] == 110
    assert N.select_as_of(res.rows, None)[0]["revenue"] == 110
    assert N.select_as_of(res.rows, date(2023, 4, 30)) == []       # nothing was public yet


def test_lower_priority_comparative_does_not_restate(app):
    """A later filing re-reporting a period with a worse concept (cash incl. restricted) must not override."""
    rows = [_fact("CashAndCashEquivalentsAtCarryingValue", 100, None, "2024-03-30", "A", "2024-05-03", fy=2024, fp="Q2"),
            _fact("CashCashEquivalentsRestrictedCashAndRestrictedCashEquivalents", 105, None, "2024-03-30", "B", "2025-05-02", fy=2025, fp="Q2")]
    res = _company(rows, fye="0930")
    bal = [r for r in res.rows if r["statement"] == "balance" and r["fiscal_quarter"] != 0]
    assert len(bal) == 1 and bal[0]["cash_and_equivalents"] == 100 and bal[0]["restated"] is False


# ------------------------------------------------------------------------------------ real fixtures
@pytest.fixture(scope="module")
def apple_rows(app):
    _, N, _, _ = _mods()
    cf = load_gz(f"companyfacts_CIK{PROTO_CIKS['AAPL']:010d}.json.gz")
    sub = load_gz(f"submissions_CIK{PROTO_CIKS['AAPL']:010d}.json.gz")
    res = N.normalize_company(320193, "AAPL", N.facts_frame(cf), N.filings_from_submissions(sub), sub["fiscalYearEnd"])
    assert res.stats["fiscal_mismatches"] == []
    return res.rows


def _row(rows, statement, fy, fq):
    return next(r for r in rows if r["statement"] == statement and r["fiscal_year"] == fy and r["fiscal_quarter"] == fq)


def test_apple_q2_fy2024_real_numbers(apple_rows):
    """Apple 10-Q filed 2024-05-03 (quarter ended 2024-03-30): the reference numbers of the spec."""
    _, N, _, _ = _mods()
    latest = N.select_as_of(apple_rows, None)
    inc = _row(latest, "income", 2024, 2)
    assert str(inc["period_end"]) == "2024-03-30" and inc["calendar_quarter"] == "2024Q1"
    assert inc["revenue"] == 90_753_000_000 and inc["net_income"] == 23_636_000_000 and inc["eps_diluted"] == 1.53
    assert inc["derived"] is False and inc["coverage"]["revenue"]["tag"] == "us-gaap:RevenueFromContractWithCustomerExcludingAssessedTax"
    bal = _row(latest, "balance", 2024, 2)
    assert bal["total_assets"] == 337_411_000_000
    cf = _row(latest, "cashflow", 2024, 2)
    assert cf["operating_cash_flow"] == 22_690_000_000            # de-cumulated: YTD6 (62,585) − Q1 (39,895)
    assert cf["derived"] is True and cf["coverage"]["operating_cash_flow"]["derived"] == "YTD6-Q1"
    q4 = _row(latest, "income", 2024, 4)
    # Apple reports 9-month YTD income facts, so Q4 = FY − YTD9 (identical to FY − (Q1+Q2+Q3), fewer inputs)
    assert q4["derived"] is True and q4["revenue"] == 94_930_000_000 and q4["coverage"]["revenue"]["derived"] == "FY-YTD9"
    fy = _row(latest, "income", 2024, 0)
    assert fy["revenue"] == 391_035_000_000 and fy["eps_diluted"] == 6.08
    # nulls are explained, never guessed
    assert inc["interest_expense"] is None and inc["coverage"]["interest_expense"]["reason"] == "no_mapped_tag"


def test_apple_ttm_and_point_in_time(apple_rows):
    _, N, _, _ = _mods()
    latest = N.select_as_of(apple_rows, None)
    t = N.ttm(latest, "income")
    w = next(x for x in t if x["fiscal_year"] == 2024 and x["fiscal_quarter"] == 2)
    assert w["revenue"] == 119_575_000_000 + 90_753_000_000 + 89_498_000_000 + 81_797_000_000 == 381_623_000_000
    assert w["ttm"] is True and w["shares_diluted"] == _row(latest, "income", 2024, 2)["shares_diluted"]
    assert w["coverage"]["eps_diluted"]["approx"]
    bt = N.ttm(latest, "balance")
    assert next(x for x in bt if x["fiscal_year"] == 2024 and x["fiscal_quarter"] == 2)["total_assets"] == 337_411_000_000
    # anti look-ahead
    assert not [r for r in N.select_as_of(apple_rows, date(2024, 5, 2)) if r["statement"] == "income" and (r["fiscal_year"], r["fiscal_quarter"]) == (2024, 2)]
    assert [r for r in N.select_as_of(apple_rows, date(2024, 5, 3)) if r["statement"] == "income" and (r["fiscal_year"], r["fiscal_quarter"]) == (2024, 2)]


def test_ratio_formulas_on_apple_q2_fy2024(apple_rows):
    """Formulas checked against the values of the recorded fixture (TTM to 2024-03-30, close 183.38 on 2024-05-03)."""
    _, N, R, _ = _mods()
    from fundamentals.service import ratio_inputs
    rows = N.select_as_of(apple_rows, date(2024, 5, 3))
    price = 183.38
    f, info = ratio_inputs(rows, price=price)
    assert str(info["period_end"]) == "2024-03-30" and info["shares_source"] == "dei:EntityCommonStockSharesOutstanding"
    values, reasons = R.compute_all(f)
    assert f["revenue"] == 381_623_000_000 and f["net_income"] == 100_389_000_000
    assert f["shares_outstanding"] == 15_334_082_000                   # cover page of the 10-Q (2024-04-19)
    assert values["market_cap"] == pytest.approx(price * 15_334_082_000)
    assert values["pe"] == pytest.approx(price / f["eps_diluted"]) and f["eps_diluted"] == pytest.approx(6.43)
    assert values["pb"] == pytest.approx(values["market_cap"] / 74_194_000_000)
    assert values["gross_margin"] == pytest.approx(173_966 / 381_623, rel=1e-6)
    assert values["net_margin"] == pytest.approx(100_389 / 381_623, rel=1e-6)
    assert values["roe"] == pytest.approx(100_389 / 74_194, rel=1e-6)
    assert values["enterprise_value"] == pytest.approx(values["market_cap"] + 104_590e6 - 32_695e6 - 34_455e6)
    assert values["ev_ebitda"] == pytest.approx(values["enterprise_value"] / f["ebitda"])
    assert values["fcf_yield"] == pytest.approx(f["free_cash_flow"] / values["market_cap"])
    assert values["current_ratio"] == pytest.approx(128_416 / 123_822, rel=1e-6)
    assert values["debt_to_equity"] == pytest.approx(104_590 / 74_194, rel=1e-6)
    assert values["forward_pe"] is None and reasons["forward_pe"] == "no_estimates"
    assert values["interest_coverage"] is None and reasons["interest_coverage"].startswith("missing:")
    assert values["revenue_growth_yoy"] == pytest.approx(381_623 / 385_095 - 1, rel=1e-6)
    assert values["book_value_ps"] == pytest.approx(74_194_000_000 / 15_334_082_000)


def test_ratio_docs_are_in_sync(app):
    _, _, R, _ = _mods()
    assert (ROOT / "docs" / "fundamentals-ratios.md").read_text(encoding="utf-8").strip() == R.render_docs().strip()
    assert len(R.RATIOS) == len(set(r.name for r in R.RATIOS)) >= 45


def test_null_ratios_never_invented(app):
    _, _, R, _ = _mods()
    values, reasons = R.compute_all({"price": 10.0, "eps_diluted": -1.0, "total_equity": 0.0, "revenue": None})
    assert values["pe"] is None and reasons["pe"] == "denominator_not_positive"
    assert values["pb"] is None and values["ps"] is None and reasons["ps"].startswith("missing:")


# -------------------------------------------------------------------------------------- filter grammar
def test_filter_grammar(app):
    from core.errors import ApiError
    _, _, _, SC = _mods()
    f = SC.parse_filters("pe<15, roe>=15%,market_cap>1.5b,ev_ebitda=5..12,exchange=Nasdaq|NYSE,ticker!=AAPL")
    assert [(x.field, x.op, x.value) for x in f] == [("pe", "<", 15.0), ("roe", ">=", 0.15), ("market_cap", ">", 1.5e9),
                                                     ("ev_ebitda", "range", (5.0, 12.0)), ("exchange", "=", ["NASDAQ", "NYSE"]),
                                                     ("ticker", "!=", ["AAPL"])]
    assert SC.parse_sort("fcf_yield:desc") == ("fcf_yield", True) and SC.parse_sort(None) == ("market_cap", True)
    for bad in ("pe<<15", "nope>1", "pe>", "pe=abc", "pe=20..10", "exchange>1", "pe~1"):
        with pytest.raises(ApiError) as e:
            SC.parse_filters(bad)
        assert e.value.code == "INVALID_FILTER"
    with pytest.raises(ApiError):
        SC.parse_sort("pe:sideways")


# ------------------------------------------------------------------------------------ edgar client
def test_token_bucket_and_backoff(app):
    import httpx
    import respx

    from fundamentals import edgar_client as ec
    calls = {"n": 0}

    def handler(request):
        calls["n"] += 1
        if calls["n"] == 1:
            return httpx.Response(429, headers={"Retry-After": "0"})
        return httpx.Response(200, json={"ok": True})

    with respx.mock(assert_all_mocked=True) as m:
        m.get("https://data.sec.gov/x.json").mock(side_effect=handler)
        c = ec.EdgarClient(rate=10_000, raw_dir=Path("/tmp/hfmd-test-raw"))
        assert c.get_json("https://data.sec.gov/x.json") == {"ok": True}
        assert c.stats.retries == 1 and calls["n"] == 2
        assert c._http.headers["User-Agent"].startswith("HF Market Data (Simon-Pierre Boucher")
    b = ec.TokenBucket(rate=1000, capacity=3)
    for _ in range(10):
        b.acquire()


def test_parse_feeds(app):
    from fundamentals import edgar_client as ec
    from fundamentals import ingest
    atom = (ROOT / "tests" / "fixtures" / "edgar" / "atom_10-Q.xml").read_text()
    entries = ec.parse_atom(atom)
    assert len(entries) == 3 and all(e["accn"] and e["cik"] and e["form"] == "10-Q" for e in entries)
    idx = ingest.parse_master_index((ROOT / "tests" / "fixtures" / "edgar" / "master.idx").read_text())
    assert idx and all(set(e) == {"cik", "name", "form", "filed", "accn"} for e in idx)
    assert ingest.master_index_url(date(2026, 9, 3)).endswith("/2026/QTR3/master.20260903.idx")