"""Tool-output eval cases — one property-assertion per behaviour that must stay true.

PR-2 Layer 2 (docs/plans/PR2_EVALUATION.md §4). These are the committed, shared
version of the fixtures Sofhia wrote during PR-0/PR-1 (which live gitignored under
`tests/`), plus one basic truth assertion for every other ACTIVE analytics tool so
the whole surface has a net.

The rule (§4): **assert a PROPERTY, not a golden float.** k-means init and ETS
optimisation have legitimate freedom, so cases assert recovered signal, correct
ordering, or bounds — never an exact fitted number. A case fails by raising
AssertionError (or any exception); the runner catches it and records the message.

NO LLM and NO DB are touched. sklearn/statsmodels fits are fine (they are not
models we call over the network) — they are what makes the ML tools' claims real.
"""

from __future__ import annotations

from collections.abc import Awaitable, Callable
from dataclasses import dataclass

import numpy as np
import pandas as pd

from src.tools.analytics.ab_test import analyze_ab_test
from src.tools.analytics.aggregation import analyze_aggregate
from src.tools.analytics.anomaly import _ANOMALY_RATE_CEILING, _flag_rate_quality, analyze_anomaly
from src.tools.analytics.cluster import analyze_cluster
from src.tools.analytics.data_quality import analyze_data_quality
from src.tools.analytics.descriptive import ColumnNotFoundError, analyze_descriptive
from src.tools.analytics.forecast import analyze_forecast
from src.tools.analytics.importance import _infer_task, analyze_importance
from src.tools.analytics.merge import analyze_merge
from src.tools.analytics.pareto import analyze_pareto
from src.tools.analytics.relationship import NonNumericColumnError, analyze_correlation
from src.tools.analytics.segmentation import analyze_segmentation
from src.tools.analytics.temporal import analyze_trend
from src.tools.analytics.visualization import render_chart


@dataclass
class EvalCase:
    """One property assertion against one tool.

    `fn` builds a fixture with a property planted, calls the tool, and asserts the
    tool reports it. It raises on failure and returns None on success.
    """

    id: str
    tool: str
    prop: str
    fn: Callable[[], Awaitable[None]]


def _assert(cond: bool, msg: str) -> None:
    if not cond:
        raise AssertionError(msg)


# --------------------------------------------------------------------------- #
# analyze_data_quality (D-1)                                                   #
# --------------------------------------------------------------------------- #
async def _dq_zero_dup_when_none() -> None:
    df = pd.DataFrame({"id": range(5), "v": [1, 2, 3, 4, 5]})  # all rows unique
    out = analyze_data_quality(df)
    _assert(out["duplicate_rows"]["count"] == 0, "unique rows must report 0 duplicates")
    _assert(out["duplicate_rows"]["rate"] == 0.0, "0 duplicates -> rate 0.0")
    _assert(out["issues"]["duplicate_row_count"] == 0, "issues must carry 0, not absent")


async def _dq_counts_planted_dups() -> None:
    df = pd.DataFrame({"a": [1, 1, 1, 2, 3], "b": ["x", "x", "x", "y", "z"]})
    out = analyze_data_quality(df)
    _assert(out["duplicate_rows"]["count"] == 2, "3 identical rows -> 2 redundant copies")
    _assert(abs(out["duplicate_rows"]["rate"] - 2 / 5) < 1e-9, "rate = 2/5")


async def _dq_subset_candidate_key() -> None:
    df = pd.DataFrame({"id": [1, 2, 3], "grp": ["A", "A", "B"]})
    out = analyze_data_quality(df, column_ids=["grp"])
    _assert(out["duplicate_rows"]["count"] == 1, "subset dup: second 'A' is one duplicate")


# --------------------------------------------------------------------------- #
# analyze_anomaly (D-3)                                                        #
# --------------------------------------------------------------------------- #
def _outlier_frame(n: int = 60) -> pd.DataFrame:
    rng = np.random.RandomState(0)
    x, y = rng.normal(0, 1, n), rng.normal(0, 1, n)
    x[:3] = [50.0, -48.0, 55.0]
    y[:3] = [50.0, -47.0, 60.0]
    return pd.DataFrame({"x": x, "y": y})


async def _an_names_rows_via_label() -> None:
    df = _outlier_frame(60)
    df["eqm_no"] = [f"UNIT-{i:03d}" for i in range(60)]
    out = await analyze_anomaly(df, ["x", "y"], contamination=0.1, label_columns=["eqm_no"])
    _assert(out["n_anomalies"] > 0, "planted outliers must be flagged")
    for rec in out["anomalies"]:
        _assert(str(rec["eqm_no"]).startswith("UNIT-"), "each record must name its row")
        _assert("x" in rec and "y" in rec, "scored values must still be present")
    named = {rec["eqm_no"] for rec in out["anomalies"]}
    _assert(bool({"UNIT-000", "UNIT-001", "UNIT-002"} & named), "planted units must appear")


async def _an_missing_label_raises() -> None:
    df = _outlier_frame(30)
    try:
        await analyze_anomaly(df, ["x", "y"], label_columns=["ghost_id"])
    except ColumnNotFoundError:
        return
    raise AssertionError("a missing label column must raise ColumnNotFoundError")


async def _an_quality_present_not_weak() -> None:
    out = await analyze_anomaly(_outlier_frame(60), ["x", "y"], contamination=0.1)
    q = out["quality"]
    _assert(q["metric"] == "anomaly_rate", "quality metric must be anomaly_rate")
    _assert(q["floor"] == _ANOMALY_RATE_CEILING, "floor must be the ceiling constant")
    _assert(q["value"] == round(out["anomaly_rate"], 4), "quality value echoes the rate")
    _assert(q["weak"] is False, "a ~0.1 flag rate is a normal run, not weak")


async def _an_flag_rate_branches() -> None:
    _assert(_flag_rate_quality(0.22)["weak"] is False, "clean-data floor 0.22 is not weak")
    _assert(_flag_rate_quality(0.50)["weak"] is False, "boundary is strict > 0.5")
    _assert(_flag_rate_quality(0.857)["weak"] is True, "0.86 flagged is weak")
    _assert("noise" in _flag_rate_quality(0.857)["note"], "weak note names it as noise")


# --------------------------------------------------------------------------- #
# analyze_importance (D-4)                                                     #
# --------------------------------------------------------------------------- #
async def _imp_infer_non_numeric() -> None:
    classify, reason = _infer_task(pd.Series(["a", "b", "a", "c"]))
    _assert(classify is True and "categor" in reason.lower(), "text target -> classification")


async def _imp_infer_float() -> None:
    classify, reason = _infer_task(pd.Series([1.5, 2.25, 3.75, 0.1]))
    _assert(classify is False and "continuous" in reason.lower(), "float target -> regression")


async def _imp_infer_ordinal_integer() -> None:
    classify, reason = _infer_task(pd.Series(list(range(15)) * 3))
    _assert(classify is False and "ordinal" in reason.lower(), "dense 15-int -> ordinal regression")


async def _imp_infer_binary_flag() -> None:
    classify, reason = _infer_task(pd.Series([0, 1, 1, 0, 1, 0]))
    _assert(classify is True and "flag" in reason.lower(), "0/1 -> binary classification")


async def _imp_infer_sparse_codes() -> None:
    classify, reason = _infer_task(pd.Series([10, 20, 700, 10, 20, 700]))
    _assert(classify is True and "categor" in reason.lower(), "sparse codes -> classification")


async def _imp_infer_many_integers() -> None:
    classify, _ = _infer_task(pd.Series(range(50)))
    _assert(classify is False, "50 distinct integers -> regression")


async def _imp_infer_all_null() -> None:
    classify, reason = _infer_task(pd.Series([None, None, None], dtype="float64"))
    _assert("no non-null" in reason.lower(), "all-null column is empty, not 'non-numeric'")


async def _imp_recovers_planted_signal() -> None:
    rng = np.random.RandomState(0)
    n = 120
    driver = rng.normal(0, 1, n)
    noise = rng.normal(0, 1, n)
    target = 5.0 * driver + 0.1 * noise + rng.normal(0, 0.05, n)
    df = pd.DataFrame({"driver": driver, "noise": noise, "target": target})
    out = await analyze_importance(df, "target", feature_columns=["driver", "noise"])
    _assert(out["task"] == "regression", "continuous target -> regression")
    top = out["features"][0]["feature"]
    _assert(top == "driver", f"planted driver must rank first, got {top}")
    _assert(out["quality"]["weak"] is False, "a strong linear signal must not be weak")


# --------------------------------------------------------------------------- #
# analyze_pareto (D-6)                                                         #
# --------------------------------------------------------------------------- #
def _revenue() -> pd.DataFrame:
    return pd.DataFrame({"region": ["A", "B", "C", "D", "A"], "revenue": [50, 10, 5, 5, 30]})


async def _pareto_additive_no_caveat() -> None:
    out = analyze_pareto(_revenue(), "region", "revenue", agg="sum")
    _assert(out["caveats"] == [], "sum on non-negative values -> no honesty caveat")
    _assert(out["categories_for_80pct"] == 1, "A alone is 80% -> categories_for_80pct == 1")


async def _pareto_non_additive_warns() -> None:
    out = analyze_pareto(_revenue(), "region", "revenue", agg="mean")
    _assert(any("not additive" in c for c in out["caveats"]), "mean must be flagged non-additive")


async def _pareto_negative_warns() -> None:
    df = pd.DataFrame({"region": ["A", "B", "C"], "delta": [50, -25, 10]})
    out = analyze_pareto(df, "region", "delta", agg="sum")
    _assert(any("negative" in c for c in out["caveats"]), "negative values must be flagged")


# --------------------------------------------------------------------------- #
# analyze_trend (D-7)                                                          #
# --------------------------------------------------------------------------- #
def _months(values: list[float]) -> pd.DataFrame:
    dates = pd.date_range("2024-01-01", periods=len(values), freq="MS")
    return pd.DataFrame({"date": dates, "v": values})


async def _trend_clean_agrees() -> None:
    out = analyze_trend(_months([10, 20, 30, 40]), "date", "v", freq="month")
    _assert(out["direction"] == "up" and out["slope"] > 0, "clean rise -> up, positive slope")
    _assert(out["r_squared"] == 1.0, "perfect line -> R^2 == 1")
    _assert(not any("Mixed signal" in c for c in out["caveats"]), "agreement -> no mixed caveat")


async def _trend_disagreement_flagged() -> None:
    out = analyze_trend(_months([100, 40, 30, 20, 10, 110]), "date", "v", freq="month")
    _assert(out["direction"] == "up", "endpoints go up (ends high)")
    _assert(out["slope"] < 0, "OLS slope goes down")
    _assert(any("Mixed signal" in c for c in out["caveats"]), "disagreement must be flagged")


# --------------------------------------------------------------------------- #
# analyze_cluster (D-10)                                                       #
# --------------------------------------------------------------------------- #
def _blobs(per: int = 12) -> pd.DataFrame:
    rng = np.random.RandomState(0)
    pts = np.vstack(
        [rng.normal(c, 0.2, (per, 2)) for c in (0, 10, 20)]
    )
    return pd.DataFrame(pts, columns=["x", "y"])


async def _cluster_override_visible() -> None:
    df = pd.DataFrame({"a": [1.0] * 25, "b": [2.0] * 25})  # identical -> collapse
    out = await analyze_cluster(df, ["a", "b"], k=5)
    _assert(out["requested_k"] == 5, "requested_k must echo the explicit k")
    _assert(out["k"] != 5, "collapse must fall back to a different k")
    _assert(any("Requested k=5" in c for c in out["caveats"]), "override must be visible")


async def _cluster_k_honoured() -> None:
    out = await analyze_cluster(_blobs(10), ["x", "y"], k=3)
    _assert(out["k"] == 3, "separable blobs -> explicit k=3 honoured")
    _assert(not any("fell back" in c for c in out["caveats"]), "no fallback on a good fit")


async def _cluster_auto_finds_3() -> None:
    out = await analyze_cluster(_blobs(12), ["x", "y"])  # no k
    _assert(out["requested_k"] is None, "auto must report requested_k None")
    _assert(out["k"] == 3, "silhouette must recover the 3 planted blobs")


# --------------------------------------------------------------------------- #
# Basic coverage — one truth assertion for every other ACTIVE tool            #
# --------------------------------------------------------------------------- #
async def _descriptive_basic() -> None:
    out = analyze_descriptive(pd.DataFrame({"revenue": [10, 20, 30, 40]}), ["revenue"])["revenue"]
    _assert(out["mean"] == 25 and out["median"] == 25, "mean/median of 10..40 = 25")
    _assert(out["min"] == 10 and out["max"] == 40, "min/max = 10/40")


async def _aggregate_groupby_sum() -> None:
    df = pd.DataFrame({"region": ["A", "A", "B", "B"], "revenue": [10, 20, 100, 200]})
    rows = analyze_aggregate(df, {"revenue": ["sum"]}, group_by=["region"])
    by = {r["region"]: r for r in rows}
    _assert(by["A"]["revenue_sum"] == 30 and by["B"]["revenue_sum"] == 300, "per-group sums")


async def _correlation_recovers_positive() -> None:
    out = analyze_correlation(pd.DataFrame({"x": [1, 2, 3, 4], "y": [2, 4, 6, 8]}), ["x", "y"])
    _assert(abs(out["matrix"]["x"]["y"] - 1.0) < 1e-9, "y=2x -> correlation 1.0")


async def _correlation_refuses_non_numeric() -> None:
    df = pd.DataFrame({"x": [1, 2], "label": ["a", "b"]})
    try:
        analyze_correlation(df, ["x", "label"])
    except NonNumericColumnError:
        return
    raise AssertionError("a non-numeric column must be refused")


async def _merge_aligns_two_metrics() -> None:
    left = pd.DataFrame({"section": ["S1", "S2"], "pa": [90.0, 80.0]})
    right = pd.DataFrame({"section": ["S1", "S2"], "backlog": [3, 7]})
    rows = analyze_merge(left, right, on=["section"], how="inner")
    by = {r["section"]: r for r in rows}
    _assert(len(rows) == 2, "inner join on matched keys -> 2 rows")
    _assert(by["S1"]["pa"] == 90.0 and by["S1"]["backlog"] == 3, "both metrics aligned per key")


def _ab(a_vals: list[float], b_vals: list[float]) -> pd.DataFrame:
    return pd.DataFrame({"grp": ["A"] * len(a_vals) + ["B"] * len(b_vals), "val": a_vals + b_vals})


async def _ab_significant_difference() -> None:
    out = analyze_ab_test(_ab([10, 11, 9, 10, 12], [20, 21, 19, 20, 22]), "grp", "val", "A", "B")
    _assert(out["comparison"] == "higher", "B > A")
    _assert(out["test"]["significant"] is True, "clearly separated groups are significant")


async def _ab_overlap_not_significant() -> None:
    out = analyze_ab_test(_ab([10, 11, 9, 10], [10, 12, 9, 11]), "grp", "val", "A", "B")
    _assert(out["test"]["significant"] is not True, "overlapping groups are not significant")


async def _segmentation_quantile_balanced() -> None:
    df = pd.DataFrame({"age": list(range(12))})
    out = analyze_segmentation(df, "age", bins=4, method="quantile")
    counts = [s["count"] for s in out["segments"]]
    _assert(len(out["segments"]) == 4, "4 quantile bins")
    _assert(max(counts) - min(counts) <= 1, "quantile bins must be balanced (counts within 1)")


async def _forecast_trend_beats_naive() -> None:
    dates = pd.date_range("2024-01-01", periods=18, freq="MS")
    df = pd.DataFrame({"month": dates, "v": [100 + 10 * i for i in range(18)]})
    out = await analyze_forecast(df, "month", "v", horizon=3, freq="month")
    _assert(len(out["forecast"]) == 3 and out["direction"] == "up", "rising trend projected up")
    q = out["quality"]
    _assert(q["value"] is not None and q["value"] < 1.0, "clean trend -> MASE < 1")
    _assert(q["weak"] is False, "a validated sub-1 MASE is not weak")


async def _forecast_short_series_not_validated() -> None:
    dates = pd.date_range("2024-01-01", periods=6, freq="MS")
    df = pd.DataFrame({"month": dates, "v": [100 + 5 * i for i in range(6)]})
    out = await analyze_forecast(df, "month", "v", horizon=2, freq="month")
    q = out["quality"]
    _assert(q["value"] is None, "too-short series cannot be validated -> no MASE number")
    _assert("not validated" in q["note"], "must say 'not validated', not a silent number")
    _assert(q["weak"] is True, "an unvalidated forecast is weak evidence")


# --------------------------------------------------------------------------- #
# render_chart — computes no numbers; its truth is a valid envelope mapping    #
# the requested columns                                                        #
# --------------------------------------------------------------------------- #
async def _chart_builds_envelope() -> None:
    df = pd.DataFrame({"region": ["A", "B", "C"], "revenue": [30, 20, 10]})
    env = render_chart(df, "bar", x="region", y="revenue")
    _assert(env["schema"] == "dataeyond.chart.v1", "must be a dataeyond.chart.v1 envelope")
    _assert(env["chart_type"] == "bar", "chart_type echoes the request")
    trace = env["plotly"]["data"][0]
    _assert(list(trace["x"]) == ["A", "B", "C"], "x maps the requested column")
    _assert(list(trace["y"]) == [30, 20, 10], "y maps the requested column")


async def _chart_missing_column_raises() -> None:
    df = pd.DataFrame({"region": ["A"], "revenue": [1]})
    try:
        render_chart(df, "bar", x="region", y="ghost")
    except ColumnNotFoundError:
        return
    raise AssertionError("a missing chart column must raise ColumnNotFoundError")


def _mk(tool: str, *fns: Callable[[], Awaitable[None]]) -> list[EvalCase]:
    """Build cases for one tool; id + prop derive from each function's name."""
    return [
        EvalCase(f.__name__.lstrip("_"), tool, f.__name__.lstrip("_").replace("_", " "), f)
        for f in fns
    ]


CASES: list[EvalCase] = [
    *_mk(
        "analyze_data_quality",
        _dq_zero_dup_when_none,
        _dq_counts_planted_dups,
        _dq_subset_candidate_key,
    ),
    *_mk(
        "analyze_anomaly",
        _an_names_rows_via_label,
        _an_missing_label_raises,
        _an_quality_present_not_weak,
        _an_flag_rate_branches,
    ),
    *_mk(
        "analyze_importance",
        _imp_infer_non_numeric,
        _imp_infer_float,
        _imp_infer_ordinal_integer,
        _imp_infer_binary_flag,
        _imp_infer_sparse_codes,
        _imp_infer_many_integers,
        _imp_infer_all_null,
        _imp_recovers_planted_signal,
    ),
    *_mk(
        "analyze_pareto",
        _pareto_additive_no_caveat,
        _pareto_non_additive_warns,
        _pareto_negative_warns,
    ),
    *_mk(
        "analyze_trend",
        _trend_clean_agrees,
        _trend_disagreement_flagged,
    ),
    *_mk(
        "analyze_cluster",
        _cluster_override_visible,
        _cluster_k_honoured,
        _cluster_auto_finds_3,
    ),
    *_mk("analyze_descriptive", _descriptive_basic),
    *_mk("analyze_aggregate", _aggregate_groupby_sum),
    *_mk(
        "analyze_correlation",
        _correlation_recovers_positive,
        _correlation_refuses_non_numeric,
    ),
    *_mk("analyze_merge", _merge_aligns_two_metrics),
    *_mk(
        "analyze_ab_test",
        _ab_significant_difference,
        _ab_overlap_not_significant,
    ),
    *_mk("analyze_segmentation", _segmentation_quantile_balanced),
    *_mk(
        "analyze_forecast",
        _forecast_trend_beats_naive,
        _forecast_short_series_not_validated,
    ),
    *_mk(
        "render_chart",
        _chart_builds_envelope,
        _chart_missing_column_raises,
    ),
]