Skip to content

Instantly share code, notes, and snippets.

@jtanx
Created November 15, 2024 23:24
Show Gist options
  • Save jtanx/27fd0c653f016e192ff7cb50decdec26 to your computer and use it in GitHub Desktop.
Save jtanx/27fd0c653f016e192ff7cb50decdec26 to your computer and use it in GitHub Desktop.
datetime chunking
import datetime as dt
from functools import partial
from typing import Any
import pandas as pd
from dateutil import relativedelta
def next_chunk(st: pd.Timestamp, et: pd.Timestamp, freqs: list[Any]):
if st == et:
return []
if not freqs:
return [(st, et)]
chunks = list(pd.date_range(st, et, freq=freqs[0](normalize=True)))
if not chunks:
return next_chunk(st, et, freqs[1:])
st_chunks = next_chunk(st, chunks[0], freqs[1:])
et_chunks = next_chunk(chunks[-1], et, freqs[1:])
res = [
*st_chunks,
*((chunks[i], chunks[i + 1]) for i in range(len(chunks) - 1)),
*et_chunks,
]
return res
if __name__ == "__main__":
stt = dt.datetime(2022, 2, 22, 15, 0, tzinfo=dt.timezone.utc)
ett = dt.datetime(2024, 11, 20, 15, 0, tzinfo=dt.timezone.utc)
qchunks = next_chunk(
pd.Timestamp(stt),
pd.Timestamp(ett),
[
pd.offsets.YearBegin,
partial(pd.offsets.QuarterBegin, startingMonth=1),
pd.offsets.MonthBegin,
],
)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment