import numpy as np import pytest from pandas._libs import lib from pandas._libs.tslibs import ( NaT, OutOfBoundsDatetime, Timedelta, Timestamp, iNaT, to_offset, ) from pandas._libs.tslibs.dtypes import NpyDatetimeUnit from pandas._libs.tslibs.period import INVALID_FREQ_ERR_MSG import pandas._testing as tm class TestTimestampRound: def test_round_division_by_zero_raises(self): ts = Timestamp("2016-01-01") msg = "Division by zero in rounding" with pytest.raises(ValueError, match=msg): ts.round("0ns") @pytest.mark.parametrize( "timestamp, freq, expected", [ ("20130101 09:10:11", "D", "20130101"), ("20130101 19:10:11", "D", "20130102"), ("20130201 12:00:00", "D", "20130202"), ("20130104 12:00:00", "D", "20130105"), ("2000-01-05 05:09:15.13", "D", "2000-01-05 00:00:00"), ("2000-01-05 05:09:15.13", "h", "2000-01-05 05:00:00"), ("2000-01-05 05:09:15.13", "s", "2000-01-05 05:09:15"), ], ) def test_round_frequencies(self, timestamp, freq, expected): dt = Timestamp(timestamp) result = dt.round(freq) expected = Timestamp(expected) assert result == expected def test_round_tzaware(self): dt = Timestamp("20130101 09:10:11", tz="US/Eastern") result = dt.round("D") expected = Timestamp("20130101", tz="US/Eastern") assert result == expected dt = Timestamp("20130101 09:10:11", tz="US/Eastern") result = dt.round("s") assert result == dt def test_round_30min(self): # round dt = Timestamp("20130104 12:32:00") result = dt.round("30Min") expected = Timestamp("20130104 12:30:00") assert result == expected def test_round_subsecond(self): # GH#14440 & GH#15578 result = Timestamp("2016-10-17 12:00:00.0015").round("ms") expected = Timestamp("2016-10-17 12:00:00.002000") assert result == expected result = Timestamp("2016-10-17 12:00:00.00149").round("ms") expected = Timestamp("2016-10-17 12:00:00.001000") assert result == expected ts = Timestamp("2016-10-17 12:00:00.0015") for freq in ["us", "ns"]: assert ts == ts.round(freq) result = Timestamp("2016-10-17 12:00:00.001501031").round("10ns") expected = Timestamp("2016-10-17 12:00:00.001501030") assert result == expected def test_round_nonstandard_freq(self): with tm.assert_produces_warning(False): Timestamp("2016-10-17 12:00:00.001501031").round("1010ns") @pytest.mark.parametrize("method", ["round", "floor", "ceil"]) @pytest.mark.parametrize( "unit, freq, freq_unit", [ ("s", "700ms", "ms"), ("s", "2500ms", "ms"), ("ms", "3us", "us"), ("us", "300ns", "ns"), ], ) def test_round_freq_not_multiple_of_resolution(self, method, unit, freq, freq_unit): # GH#67978 freq is neither a multiple nor a divisor of one unit of # self's resolution, so the result would not be representable. # The value is deliberately off the freq grid so the second half # below is not a no-op for any of the parametrizations. ts = Timestamp("2020-01-01 00:00:06.500000001") msg = rf"freq=.* is incompatible with unit={unit}" with pytest.raises(ValueError, match=msg): getattr(ts.as_unit(unit), method)(freq) finer = ts.as_unit(freq_unit) result = getattr(finer, method)(freq) assert result != finer assert result == getattr(finer.as_unit("ns"), method)(freq) @pytest.mark.parametrize("method", ["round", "floor", "ceil"]) @pytest.mark.parametrize("unit, freq", [("s", "250ms"), ("us", "250ns")]) def test_round_freq_divides_resolution(self, method, unit, freq): # GH#67978 freq divides one unit of self's resolution evenly, so every # representable value is already a multiple of freq ts = Timestamp("2020-01-01 00:00:06").as_unit(unit) assert getattr(ts, method)(freq) == ts def test_round_invalid_arg(self): stamp = Timestamp("2000-01-05 05:09:15.13") with pytest.raises(ValueError, match=INVALID_FREQ_ERR_MSG): stamp.round("foo") @pytest.mark.parametrize( "timestamp,frequency,expected_ceil,expected_round,expected_floor", [ ( Timestamp("20130101 09:10:11"), Timedelta("1 day"), Timestamp("20130102"), Timestamp("20130101"), Timestamp("20130101"), ), ( Timestamp("20130101 19:10:11"), Timedelta("1 day"), Timestamp("20130102"), Timestamp("20130102"), Timestamp("20130101"), ), ( Timestamp("2000-01-05 05:09:15.13"), Timedelta("1 day"), Timestamp("2000-01-06 00:00:00"), Timestamp("2000-01-05 00:00:00"), Timestamp("2000-01-05 00:00:00"), ), ( Timestamp("2000-01-05 05:09:15.13"), Timedelta("1 hour"), Timestamp("2000-01-05 06:00:00"), Timestamp("2000-01-05 05:00:00"), Timestamp("2000-01-05 05:00:00"), ), ( Timestamp("2000-01-05 05:09:15.13"), Timedelta("1 second"), Timestamp("2000-01-05 05:09:16"), Timestamp("2000-01-05 05:09:15"), Timestamp("2000-01-05 05:09:15"), ), ( Timestamp("2000-01-05 05:09:15.13"), Timedelta("1 hour 30 min"), Timestamp("2000-01-05 06:00:00"), Timestamp("2000-01-05 04:30:00"), Timestamp("2000-01-05 04:30:00"), ), # Edge cases derived from TestTimestampRound.test_ceil_floor_edge ( Timestamp("2117-01-01 00:00:45"), Timedelta("15s"), Timestamp("2117-01-01 00:00:45"), Timestamp("2117-01-01 00:00:45"), Timestamp("2117-01-01 00:00:45"), ), ( Timestamp("2117-01-01 00:00:45.000000012"), Timedelta("10ns"), Timestamp("2117-01-01 00:00:45.000000020"), Timestamp("2117-01-01 00:00:45.000000010"), Timestamp("2117-01-01 00:00:45.000000010"), ), ( Timestamp("1823-01-01 00:00:01"), Timedelta("1s"), Timestamp("1823-01-01 00:00:01"), Timestamp("1823-01-01 00:00:01"), Timestamp("1823-01-01 00:00:01"), ), ( Timestamp("1823-01-01 00:00:01.000000012"), Timedelta("10ns"), Timestamp("1823-01-01 00:00:01.000000020"), Timestamp("1823-01-01 00:00:01.000000010"), Timestamp("1823-01-01 00:00:01.000000010"), ), ], ) def test_rounding_with_timedelta_freq( self, timestamp, frequency, expected_ceil, expected_round, expected_floor ): # GH#63687 - Timestamp rounding methods accept Timedelta arguments assert timestamp.ceil(frequency) == expected_ceil assert timestamp.round(frequency) == expected_round assert timestamp.floor(frequency) == expected_floor def test_rounding_nat_frequency(self): ts = Timestamp("2000-01-05 05:09:15.13") with pytest.raises(TypeError, match="Argument 'freq' has incorrect type"): ts.ceil(NaT) with pytest.raises(TypeError, match="Argument 'freq' has incorrect type"): ts.floor(NaT) with pytest.raises(TypeError, match="Argument 'freq' has incorrect type"): ts.round(NaT) def test_rounding_nat_timestamp(self): freq = Timedelta("1s") assert NaT.ceil(freq) is NaT assert NaT.floor(freq) is NaT assert NaT.round(freq) is NaT @pytest.mark.parametrize( "test_input, rounder, freq, expected", [ ("2117-01-01 00:00:45", "floor", "15s", "2117-01-01 00:00:45"), ("2117-01-01 00:00:45", "ceil", "15s", "2117-01-01 00:00:45"), ( "2117-01-01 00:00:45.000000012", "floor", "10ns", "2117-01-01 00:00:45.000000010", ), ( "1823-01-01 00:00:01.000000012", "ceil", "10ns", "1823-01-01 00:00:01.000000020", ), ("1823-01-01 00:00:01", "floor", "1s", "1823-01-01 00:00:01"), ("1823-01-01 00:00:01", "ceil", "1s", "1823-01-01 00:00:01"), ("NaT", "floor", "1s", "NaT"), ("NaT", "ceil", "1s", "NaT"), ], ) def test_ceil_floor_edge(self, test_input, rounder, freq, expected): dt = Timestamp(test_input) func = getattr(dt, rounder) result = func(freq) if dt is NaT: assert result is NaT else: expected = Timestamp(expected) assert result == expected @pytest.mark.parametrize( "test_input, freq, expected", [ ("2018-01-01 00:02:06", "2s", "2018-01-01 00:02:06"), ("2018-01-01 00:02:00", "2min", "2018-01-01 00:02:00"), ("2018-01-01 00:04:00", "4min", "2018-01-01 00:04:00"), ("2018-01-01 00:15:00", "15min", "2018-01-01 00:15:00"), ("2018-01-01 00:20:00", "20min", "2018-01-01 00:20:00"), ("2018-01-01 03:00:00", "3h", "2018-01-01 03:00:00"), ], ) @pytest.mark.parametrize("rounder", ["ceil", "floor", "round"]) def test_round_minute_freq(self, test_input, freq, expected, rounder): # Ensure timestamps that shouldn't round dont! # GH#21262 dt = Timestamp(test_input) expected = Timestamp(expected) func = getattr(dt, rounder) result = func(freq) assert result == expected def test_ceil(self, unit): dt = Timestamp("20130101 09:10:11").as_unit(unit) result = dt.ceil("D") expected = Timestamp("20130102") assert result == expected assert result._creso == dt._creso def test_floor(self, unit): dt = Timestamp("20130101 09:10:11").as_unit(unit) result = dt.floor("D") expected = Timestamp("20130101") assert result == expected assert result._creso == dt._creso @pytest.mark.parametrize("method", ["ceil", "round", "floor"]) def test_round_dst_border_ambiguous(self, method, unit): # GH 18946 round near "fall back" DST ts = Timestamp("2017-10-29 00:00:00", tz="UTC").tz_convert("Europe/Madrid") ts = ts.as_unit(unit) result = getattr(ts, method)("h", ambiguous=True) assert result == ts assert result._creso == getattr(NpyDatetimeUnit, f"NPY_FR_{unit}").value result = getattr(ts, method)("h", ambiguous=False) expected = Timestamp("2017-10-29 01:00:00", tz="UTC").tz_convert( "Europe/Madrid" ) assert result == expected assert result._creso == getattr(NpyDatetimeUnit, f"NPY_FR_{unit}").value result = getattr(ts, method)("h", ambiguous="NaT") assert result is NaT msg = "Cannot infer dst time" with pytest.raises(ValueError, match=msg): getattr(ts, method)("h", ambiguous="raise") @pytest.mark.parametrize( "method, ts_str, freq", [ ["ceil", "2018-03-11 01:59:00-0600", "5min"], ["round", "2018-03-11 01:59:00-0600", "5min"], ["floor", "2018-03-11 03:01:00-0500", "2h"], ], ) def test_round_dst_border_nonexistent(self, method, ts_str, freq, unit): # GH 23324 round near "spring forward" DST ts = Timestamp(ts_str, tz="America/Chicago").as_unit(unit) result = getattr(ts, method)(freq, nonexistent="shift_forward") expected = Timestamp("2018-03-11 03:00:00", tz="America/Chicago") assert result == expected assert result._creso == getattr(NpyDatetimeUnit, f"NPY_FR_{unit}").value result = getattr(ts, method)(freq, nonexistent="NaT") assert result is NaT msg = "2018-03-11 02:00:00" with pytest.raises(ValueError, match=msg): getattr(ts, method)(freq, nonexistent="raise") @pytest.mark.parametrize( "timestamp", [ "2018-01-01 0:0:0.124999360", "2018-01-01 0:0:0.125000367", "2018-01-01 0:0:0.125500", "2018-01-01 0:0:0.126500", "2018-01-01 12:00:00", "2019-01-01 12:00:00", ], ) @pytest.mark.parametrize( "freq", [ "2ns", "3ns", "4ns", "5ns", "6ns", "7ns", "250ns", "500ns", "750ns", "1us", "19us", "250us", "500us", "750us", "1s", "2s", "3s", "1D", ], ) def test_round_int64(self, timestamp, freq): # check that all rounding modes are accurate to int64 precision # see GH#22591 dt = Timestamp(timestamp).as_unit("ns") unit = to_offset(freq).nanos # test floor result = dt.floor(freq) assert result._value % unit == 0, f"floor not a {freq} multiple" assert 0 <= dt._value - result._value < unit, "floor error" # test ceil result = dt.ceil(freq) assert result._value % unit == 0, f"ceil not a {freq} multiple" assert 0 <= result._value - dt._value < unit, "ceil error" # test round result = dt.round(freq) assert result._value % unit == 0, f"round not a {freq} multiple" assert abs(result._value - dt._value) <= unit // 2, "round error" if unit % 2 == 0 and abs(result._value - dt._value) == unit // 2: # round half to even assert result._value // unit % 2 == 0, "round half to even error" def test_round_implementation_bounds(self): # See also: analogous test for Timedelta result = Timestamp.min.ceil("s") expected = Timestamp(1677, 9, 21, 0, 12, 44) assert result == expected result = Timestamp.max.floor("s") expected = Timestamp.max - Timedelta(854775807) assert result == expected msg = "Cannot round 1677-09-21 00:12:43.145224193 to freq=" with pytest.raises(OutOfBoundsDatetime, match=msg): Timestamp.min.floor("s") with pytest.raises(OutOfBoundsDatetime, match=msg): Timestamp.min.round("s") msg = "Cannot round 2262-04-11 23:47:16.854775807 to freq=" with pytest.raises(OutOfBoundsDatetime, match=msg): Timestamp.max.ceil("s") with pytest.raises(OutOfBoundsDatetime, match=msg): Timestamp.max.round("s") @pytest.mark.parametrize( "val", [ iNaT + 1, -1, 0, 1, lib.i8max, 10**9 - 1, 10**9, 10**9 + 1, 60 * 10**9 - 1, 24 * 3600 * 10**9 - 1, -(10**9 + 1), -(60 * 10**9 - 1), -(24 * 3600 * 10**9 - 1), ], ) @pytest.mark.parametrize( "method", [Timestamp.round, Timestamp.floor, Timestamp.ceil] ) def test_round_sanity(self, val, method): err_cls = OutOfBoundsDatetime val = np.int64(val) ts = Timestamp(val) def checker(ts, nanos, unit): # First check that we do raise in cases where we should if nanos == 1: pass else: div, mod = divmod(ts._value, nanos) diff = int(nanos - mod) lb = ts._value - mod assert lb <= ts._value # i.e. no overflows with python ints ub = ts._value + diff assert ub > ts._value # i.e. no overflows with python ints msg = "without overflow" if mod == 0: # We should never be raising in this pass elif method is Timestamp.ceil: if ub > Timestamp.max._value: with pytest.raises(err_cls, match=msg): method(ts, unit) return elif method is Timestamp.floor: if lb < Timestamp.min._value: with pytest.raises(err_cls, match=msg): method(ts, unit) return elif mod >= diff: if ub > Timestamp.max._value: with pytest.raises(err_cls, match=msg): method(ts, unit) return elif lb < Timestamp.min._value: with pytest.raises(err_cls, match=msg): method(ts, unit) return res = method(ts, unit) td = res - ts diff = abs(td._value) assert diff < nanos assert res._value % nanos == 0 if method is Timestamp.round: assert diff <= nanos / 2 elif method is Timestamp.floor: assert res <= ts elif method is Timestamp.ceil: assert res >= ts nanos = 1 checker(ts, nanos, "ns") nanos = 1000 checker(ts, nanos, "us") nanos = 1_000_000 checker(ts, nanos, "ms") nanos = 1_000_000_000 checker(ts, nanos, "s") nanos = 60 * 1_000_000_000 checker(ts, nanos, "min") nanos = 60 * 60 * 1_000_000_000 checker(ts, nanos, "h") nanos = 24 * 60 * 60 * 1_000_000_000 checker(ts, nanos, "D")