Last active
July 12, 2021 08:02
-
-
Save cjw296/73af04f7a4dcbc39f2ec01384bf5aa90 to your computer and use it in GitHub Desktop.
pd.date_range for 9am each day between two pd.Timestamp objects
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import unittest | |
from datetime import time | |
from typing import Callable, Iterable | |
import pandas as pd | |
from pandas._libs.tslibs import to_offset | |
class Tests: | |
implementation: Callable[[pd.Timestamp, pd.Timestamp], Iterable[pd.Timestamp]] | |
assertEqual: Callable[[Iterable[pd.Timestamp], Iterable[pd.Timestamp]], None] | |
def check(self, result, expected): | |
self.assertEqual(list(result), list(pd.DatetimeIndex(expected))) | |
def test_start_before_end_after(self): | |
result = self.implementation( | |
pd.Timestamp('2021-07-05 08:10:11'), | |
pd.Timestamp('2021-07-07 12:13:14') | |
) | |
self.check(result, ['2021-07-05 09:00:00', '2021-07-06 09:00:00', '2021-07-07 09:00:00']) | |
def test_start_after_end_before(self): | |
result = self.implementation( | |
pd.Timestamp('2021-07-05 09:10:11'), | |
pd.Timestamp('2021-07-07 08:13:14') | |
) | |
self.check(result, ['2021-07-06 09:00:00']) | |
def test_all_datetime_bits(self): | |
result = self.implementation( | |
pd.Timestamp(2021, 7, 5, 8, 10, 11, 1234), | |
pd.Timestamp(2021, 7, 7, 12, 13, 14, 1516), | |
) | |
self.check(result, ['2021-07-05 09:00:00', '2021-07-06 09:00:00', '2021-07-07 09:00:00']) | |
def test_weekend(self): | |
result = self.implementation( | |
pd.Timestamp('2021-07-02 08:10:11'), | |
pd.Timestamp('2021-07-04 09:13:14') | |
) | |
self.check(result, ['2021-07-02 09:00:00', '2021-07-03 09:00:00', '2021-07-04 09:00:00']) | |
def test_one_second_after_nine(self): | |
result = self.implementation( | |
pd.Timestamp('2021-07-04 09:00:01'), | |
pd.Timestamp('2021-07-07 09:35:44.964864') | |
) | |
self.check(result, ['2021-07-05 09:00:00', '2021-07-06 09:00:00', '2021-07-07 09:00:00']) | |
def test_closed_endpoints(self): | |
result = self.implementation( | |
pd.Timestamp('2021-07-05 09:00:00'), | |
pd.Timestamp('2021-07-07 09:00:00') | |
) | |
self.check(result, ['2021-07-05 09:00:00', '2021-07-06 09:00:00', '2021-07-07 09:00:00']) | |
class TestMethod1(Tests, unittest.TestCase): | |
# Erfan, Ben | |
@staticmethod | |
def implementation(start, end): | |
return pd.date_range(pd.Timestamp.combine(start.date(), time(9)), | |
pd.Timestamp.combine(end.date(), time(9))) | |
class TestMethod2(Tests, unittest.TestCase): | |
# Erfan | |
@staticmethod | |
def implementation(start, end): | |
dates = pd.date_range(start, end, freq=pd.offsets.BusinessHour()).to_series() | |
return dates.groupby(dates.dt.date).min() | |
class TestMethod3(Tests, unittest.TestCase): | |
# Tomasz | |
@staticmethod | |
def implementation(start, end): | |
return pd.date_range(*(d.floor('1d') + pd.Timedelta('9h') for d in (start, end)), freq='d') | |
class TestMethod4(Tests, unittest.TestCase): | |
# Dimi | |
@staticmethod | |
def implementation(start, end): | |
return [x for x in pd.date_range(start.round('D') + pd.Timedelta(hours=9), end, freq='1D') | |
if start <= x <= end] | |
class TestMethod4A(Tests, unittest.TestCase): | |
# Matt | |
@staticmethod | |
def implementation(start, end, offset = pd.Timedelta(hours=9)): | |
return [x + offset for x in pd.date_range(start, end, normalize=True) | |
if start <= x + offset <= end] | |
class TestMethod5(Tests, unittest.TestCase): | |
# Andrew | |
@staticmethod | |
def implementation(start, end): | |
df = pd.DataFrame(index=[start, end]).resample("1D").asfreq() | |
df.index = df.index + to_offset("9h") | |
return df.index | |
class TestMethod6(Tests, unittest.TestCase): | |
# Nakul | |
@staticmethod | |
def implementation(start, end, offset=pd.Timedelta("9h")): | |
endpoints = [(date - offset).normalize() for date in (start, end)] | |
return pd.date_range(*endpoints, closed="right") + pd.offsets.Hour(9) | |
class TestMethod7(Tests, unittest.TestCase): | |
# Joris | |
@staticmethod | |
def implementation(start, end, offset=pd.Timedelta("9h")): | |
idx = pd.date_range(start, end, normalize=True) + offset | |
idx = idx[(idx >= start) & (idx <= end)] | |
return idx | |
if __name__ == '__main__': | |
unittest.main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
$ python ../datetime_range.py -q 2>&1 | grep FAIL: | |
FAIL: test_one_second_after_nine (__main__.TestMethod1) | |
FAIL: test_start_after_end_before (__main__.TestMethod1) | |
FAIL: test_start_after_end_before (__main__.TestMethod2) | |
FAIL: test_weekend (__main__.TestMethod2) | |
FAIL: test_one_second_after_nine (__main__.TestMethod3) | |
FAIL: test_start_after_end_before (__main__.TestMethod3) | |
FAIL: test_one_second_after_nine (__main__.TestMethod5) | |
FAIL: test_start_after_end_before (__main__.TestMethod5) | |
FAIL: test_closed_endpoints (__main__.TestMethod6) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
This is also written up on stackoverflow and as a pandas feature request.