import numpy as np
import pytest

from pandas.compat import HAS_PYARROW

import pandas as pd
import pandas._testing as tm


@pytest.mark.parametrize("operation, expected", [("min", "a"), ("max", "b")])
def test_reductions_series_strings(operation, expected):
    # GH#31746
    ser = pd.Series(["a", "b"], dtype="string")
    res_operation_serie = getattr(ser, operation)()
    assert res_operation_serie == expected


@pytest.mark.parametrize("as_period", [True, False])
def test_mode_extension_dtype(as_period):
    # GH#41927 preserve dt64tz dtype
    ser = pd.Series([pd.Timestamp(1979, 4, n) for n in range(1, 5)])

    if as_period:
        ser = ser.dt.to_period("D")
    else:
        ser = ser.dt.tz_localize("US/Central")

    res = ser.mode()
    assert res.dtype == ser.dtype
    tm.assert_series_equal(res, ser)


def test_mode_nullable_dtype(any_numeric_ea_dtype):
    # GH#55340
    ser = pd.Series([1, 3, 2, pd.NA, 3, 2, pd.NA], dtype=any_numeric_ea_dtype)
    result = ser.mode(dropna=False)
    expected = pd.Series([2, 3, pd.NA], dtype=any_numeric_ea_dtype)
    tm.assert_series_equal(result, expected)

    result = ser.mode(dropna=True)
    expected = pd.Series([2, 3], dtype=any_numeric_ea_dtype)
    tm.assert_series_equal(result, expected)

    ser[-1] = pd.NA

    result = ser.mode(dropna=True)
    expected = pd.Series([2, 3], dtype=any_numeric_ea_dtype)
    tm.assert_series_equal(result, expected)

    result = ser.mode(dropna=False)
    expected = pd.Series([pd.NA], dtype=any_numeric_ea_dtype)
    tm.assert_series_equal(result, expected)


def test_mode_nullable_dtype_edge_case(any_numeric_ea_dtype):
    # GH##58926
    ser = pd.Series([1, 2, 3, 1], dtype=any_numeric_ea_dtype)
    result = ser.mode(dropna=False)
    expected = pd.Series([1], dtype=any_numeric_ea_dtype)
    tm.assert_series_equal(result, expected)

    ser2 = pd.Series([1, 1, 2, 3, pd.NA], dtype=any_numeric_ea_dtype)
    result = ser2.mode(dropna=False)
    expected = pd.Series([1], dtype=any_numeric_ea_dtype)
    tm.assert_series_equal(result, expected)

    ser3 = pd.Series([1, pd.NA, pd.NA], dtype=any_numeric_ea_dtype)
    result = ser3.mode(dropna=False)
    expected = pd.Series([pd.NA], dtype=any_numeric_ea_dtype)
    tm.assert_series_equal(result, expected)

    ser4 = pd.Series([1, 1, pd.NA, pd.NA], dtype=any_numeric_ea_dtype)
    result = ser4.mode(dropna=False)
    expected = pd.Series([1, pd.NA], dtype=any_numeric_ea_dtype)
    tm.assert_series_equal(result, expected)


def test_mode_string(any_string_dtype):
    # GH#56183
    ser = pd.Series(["a", "b"], dtype=any_string_dtype)
    result = ser.mode()
    expected = pd.Series(["a", "b"], dtype=any_string_dtype)
    tm.assert_series_equal(result, expected)


def test_reductions_td64_with_nat():
    # GH#8617
    ser = pd.Series([0, pd.NaT], dtype="m8[ns]")
    exp = ser[0]
    assert ser.median() == exp
    assert ser.min() == exp
    assert ser.max() == exp


def test_td64_sum_empty(skipna):
    # GH#37151
    ser = pd.Series([], dtype="timedelta64[ns]")

    result = ser.sum(skipna=skipna)
    assert isinstance(result, pd.Timedelta)
    assert result == pd.Timedelta(0)


def test_td64_summation_overflow():
    # GH#9442
    ser = pd.Series(pd.date_range("20130101", periods=100000, freq="h", unit="ns"))
    ser[0] += pd.Timedelta("1s 1ms")

    # mean
    result = (ser - ser.min()).mean()
    expected = pd.Timedelta((pd.TimedeltaIndex(ser - ser.min()).asi8 / len(ser)).sum())

    # the computation is converted to float so
    # might be some loss of precision
    assert np.allclose(result._value / 1000, expected._value / 1000)

    # sum
    # GH#43178: OutOfBoundsTimedelta (a ValueError subclass) is raised
    msg = "overflow in timedelta operation"
    with pytest.raises(pd.errors.OutOfBoundsTimedelta, match=msg):
        (ser - ser.min()).sum()

    s1 = ser[0:10000]
    with pytest.raises(pd.errors.OutOfBoundsTimedelta, match=msg):
        (s1 - s1.min()).sum()
    s2 = ser[0:1000]
    (s2 - s2.min()).sum()


def test_td64_sum_all_nat_skipna_false():
    # GH#43178: an all-NaT sum with skipna=False must return NaT rather than be
    #  mistaken for an overflow (the NaT sentinels accumulate past int64 bounds)
    ser = pd.Series(np.array(["NaT", "NaT", "NaT"], dtype="m8[ns]"))
    assert ser.sum(skipna=False) is pd.NaT


def test_td64_sum_exact():
    # GH#66551: the sum used to accumulate in float64, whose 53-bit mantissa
    #  silently rounded results above 2**53
    ser = pd.Series([pd.Timedelta(2**53 + 1, "ns"), pd.Timedelta(0, "ns")])
    assert ser.sum()._value == 2**53 + 1

    # a single representable value has to survive the round trip
    ser = pd.Series([pd.Timedelta.min])
    assert ser.sum() == pd.Timedelta.min

    # operands that individually round but cancel exactly
    ser = pd.Series([pd.Timedelta(2**62 + 1, "ns"), pd.Timedelta(-(2**62), "ns")])
    assert ser.sum() == pd.Timedelta(1, "ns")


def test_td64_sum_on_nat_sentinel():
    # GH#66551: a total of exactly int64.min is representable but
    #  indistinguishable from NaT once stored
    ser = pd.Series([pd.Timedelta.min, pd.Timedelta(-1, "ns")])
    with pytest.raises(pd.errors.OutOfBoundsTimedelta, match="overflow"):
        ser.sum()


def test_td64_sum_nat_positions_exempt():
    # GH#66551: with skipna=False the result is NaT anyway, so an overflow among
    #  the entries the NaT masks must not raise
    ser = pd.Series([pd.Timedelta.max, pd.NaT, pd.Timedelta.max])
    assert ser.sum(skipna=False) is pd.NaT


def test_td64_sum_below_min_count_exempt():
    # GH#66551: min_count nulls the result out, so it must not raise
    ser = pd.Series([pd.Timedelta.max] * 2)
    assert ser.sum(min_count=5) is pd.NaT


def test_prod_numpy16_bug():
    ser = pd.Series([1.0, 1.0, 1.0], index=range(3))
    result = ser.prod()

    assert not isinstance(result, pd.Series)


@pytest.mark.parametrize("func", [np.any, np.all])
@pytest.mark.parametrize("kwargs", [{"keepdims": True}, {"out": object()}])
def test_validate_any_all_out_keepdims_raises(kwargs, func):
    ser = pd.Series([1, 2])
    param = next(iter(kwargs))
    name = func.__name__

    msg = (
        f"the '{param}' parameter is not "
        "supported in the pandas "
        rf"implementation of {name}\(\)"
    )
    with pytest.raises(ValueError, match=msg):
        func(ser, **kwargs)


def test_validate_sum_initial():
    ser = pd.Series([1, 2])
    msg = (
        r"the 'initial' parameter is not "
        r"supported in the pandas "
        r"implementation of sum\(\)"
    )
    with pytest.raises(ValueError, match=msg):
        np.sum(ser, initial=10)


def test_validate_median_initial():
    ser = pd.Series([1, 2])
    msg = (
        r"the 'overwrite_input' parameter is not "
        r"supported in the pandas "
        r"implementation of median\(\)"
    )
    with pytest.raises(ValueError, match=msg):
        # It seems like np.median doesn't dispatch, so we use the
        # method instead of the ufunc.
        ser.median(overwrite_input=True)


def test_validate_stat_keepdims():
    ser = pd.Series([1, 2])
    msg = (
        r"the 'keepdims' parameter is not "
        r"supported in the pandas "
        r"implementation of sum\(\)"
    )
    with pytest.raises(ValueError, match=msg):
        np.sum(ser, keepdims=True)


def test_mean_with_convertible_string_raises(using_infer_string):
    # GH#44008
    ser = pd.Series(["1", "2"])
    assert ser.sum() == "12"

    # str dtype vs object dtype (infer_string=False) raise differently; drop the
    # object-dtype branch once infer_string=False is no longer supported.
    msg = (
        "Cannot perform reduction 'mean' with string dtype"
        if using_infer_string
        else "Could not convert '1' to numeric"
    )
    with pytest.raises(TypeError, match=msg):
        ser.mean()

    with pytest.raises(TypeError, match=msg):
        ser.to_frame().mean()


def test_mean_dont_convert_j_to_complex(using_infer_string):
    # GH#36703
    df = pd.DataFrame([{"db": "J", "numeric": 123}])
    # str dtype vs object dtype (infer_string=False) raise differently; drop the
    # object-dtype branches once infer_string=False is no longer supported.
    msg = (
        "Cannot perform reduction 'mean' with string dtype"
        if using_infer_string
        else "Could not convert 'J' to numeric"
    )
    with pytest.raises(TypeError, match=msg):
        df.mean()

    with pytest.raises(TypeError, match=msg):
        df.agg("mean")

    # pyarrow-backed str raises the string-dtype reduction error; object dtype
    # converts. Without pyarrow the inferred str dtype can fall back either way
    # (depends on construction), so allow both there.
    if using_infer_string and HAS_PYARROW:
        msg = "Cannot perform reduction 'mean' with string dtype"
    else:
        msg = "|".join(
            [
                "Could not convert 'J' to numeric",
                "Cannot perform reduction 'mean' with string dtype",
            ]
        )
    with pytest.raises(TypeError, match=msg):
        df["db"].mean()

    # .astype("string") forces str dtype regardless of the infer_string setting
    msg = "Cannot perform reduction 'mean' with string dtype"
    with pytest.raises(TypeError, match=msg):
        np.mean(df["db"].astype("string").array)


def test_median_with_convertible_string_raises(using_infer_string):
    # GH#34671 this _could_ return a string "2", but definitely not float 2.0
    # str dtype vs object dtype (infer_string=False) raise differently; drop the
    # object-dtype branch once infer_string=False is no longer supported.
    msg = (
        "Cannot perform reduction 'median' with string dtype"
        if using_infer_string
        else "Could not convert '1' to numeric"
    )
    ser = pd.Series(["1", "2", "3"])
    with pytest.raises(TypeError, match=msg):
        ser.median()

    with pytest.raises(TypeError, match=msg):
        ser.to_frame().median()