Skip to content
This repository has been archived by the owner on Sep 4, 2024. It is now read-only.

Commit

Permalink
consistent output in FedIndex to DatetimeIndex for date attrs
Browse files Browse the repository at this point in the history
  • Loading branch information
seekinginfiniteloop committed Dec 27, 2023
1 parent eb99d37 commit 7cdf9e0
Show file tree
Hide file tree
Showing 4 changed files with 73 additions and 103 deletions.
33 changes: 16 additions & 17 deletions fedcal/_date_attributes.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,9 @@
from __future__ import annotations

import warnings
from typing import ClassVar, Generator
from typing import ClassVar

import numpy as np
import pandas as pd
from attrs import define, field
from numpy.typing import NDArray
Expand Down Expand Up @@ -68,7 +69,8 @@ def __attrs_post_init__(self) -> None:
if not type(self).fed_business_days:
hol_instance = FedHolidays()
type(self).fed_business_days = CustomBusinessDay(
normalize=True, calendar=hol_instance.holidays
normalize=True,
holidays=hol_instance.holidays,
)

@classmethod
Expand All @@ -91,27 +93,19 @@ def get_business_days(
dates: DatetimeIndex = ensure_datetimeindex(dates=dates)
with warnings.catch_warnings():
warnings.simplefilter(action="ignore")
next_business_days = dates + cls.fed_business_days
return next_business_days == dates
return dates.isin(values=(dates+cls.fed_business_days))

def get_prior_business_day(
self, date: Timestamp
) -> Generator[Timestamp, None, None]:
def get_prior_business_day(self, date: Timestamp) -> Timestamp:
"""
Generates next earliest business day. Primarily for finding
next-earliest business day before a military payday that doesn't
fall on a business day.
Yields
------
Returns
-------
next nearest business day prior to the given date
"""
current_day: Timestamp = date - CustomBusinessDay(
calendar=self.fed_business_days.calendar
)
while current_day < date:
yield current_day
current_day += CustomBusinessDay(calendar=self.fed_business_days.calendar)
return self.fed_business_days.rollback(dt=date)


@define(order=True, auto_attribs=True)
Expand Down Expand Up @@ -218,8 +212,13 @@ def guess_proclamation_holidays(
)
christmas_eves = christmas_days - pd.DateOffset(normalize=True, days=1)
filtered_dates: DatetimeIndex = dates[dates.year > 2023]
return christmas_days.weekday.isin(values=[1, 4]) & (
filtered_dates == christmas_eves
return (
None
if filtered_dates.empty
else (
christmas_days.weekday.isin(values=[1, 4])
& (filtered_dates == christmas_eves)
)
)


Expand Down
5 changes: 4 additions & 1 deletion fedcal/_dept_status.py
Original file line number Diff line number Diff line change
Expand Up @@ -396,4 +396,7 @@ def get_state_for_range_generator(
last_known_status[department] = status_pool[
(department, status_key)
]
yield (str(key_date), last_known_status)
yield (
time_utils.to_timestamp(key_date).strftime(format="%Y-%m-%d"),
last_known_status,
)
61 changes: 17 additions & 44 deletions fedcal/_mil.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,40 +79,21 @@ def get_mil_paydays(
"""

dates = pd.DatetimeIndex(data=self.dates if dates is None else dates)
bizday_instance = _date_attributes.FedBusDay()

first_or_fifteenth_mask: Series = pd.Series(
data=(dates.day.isin(values=[1, 15])), index=dates, dtype=bool
)

business_days_mask = pd.Series(
data=dates.isin(
values=bizday_instance.get_business_days(dates=dates).index
),
bday = _date_attributes.FedBusDay()
bdays = pd.Series(data=(bday.get_business_days(dates=dates)), index=dates)
pays = pd.Series(
data=(dates.day.isin(values=[1, 15]) & bdays),
index=dates,
dtype=bool,
name="mil_paydays",
)

payday_mask: Series[bool] = first_or_fifteenth_mask & business_days_mask

non_biz_1st_15th: DatetimeIndex = dates[
first_or_fifteenth_mask & ~business_days_mask
]
for non_biz_date in non_biz_1st_15th:
prev_days: DatetimeIndex = pd.date_range(
start=non_biz_date - pd.Timedelta(days=3),
end=non_biz_date - pd.Timedelta(days=1),
non_std: DatetimeIndex = dates[dates.day.isin(values=[1, 15]) & ~pays]
for non_std_date in non_std:
closest_bday: Timestamp = bday.get_prior_business_day(
date=non_std_date - pd.Timedelta(days=1)
)
prev_business_days: bool | pd.Series[
bool
] = bizday_instance.get_business_days(dates=prev_days)
prev_business_day_mask: NDArray = prev_days.isin(values=prev_business_days)
recent_biz_day: Timestamp = prev_days.to_series()[
prev_business_day_mask
].max()
payday_mask.at[recent_biz_day] = recent_biz_day in dates

return payday_mask
if closest_bday in dates:
pays.at[closest_bday] = True
return pays


@define(order=True, kw_only=True)
Expand Down Expand Up @@ -213,7 +194,7 @@ def get_probable_passdays(
dates[masks["wednesday_holidays"]] + pd.DateOffset(days=1)
)

return (fri_mask | mon_mask | thurs_mask) & masks["eligible_days"]
return (fri_mask | mon_mask | thurs_mask) & masks["business_days"]

@staticmethod
def _get_base_masks(
Expand All @@ -233,28 +214,20 @@ def _get_base_masks(
A dataframe of boolean masks, and date information
"""

bizday_instance = _date_attributes.FedBusDay()
holidays_instance = _date_attributes.FedHolidays()
bday = _date_attributes.FedBusDay()
holiday = _date_attributes.FedHolidays()

mask_frame = pd.DataFrame(index=dates)

mask_frame["dates"] = dates.to_series(index=dates)
mask_frame["holidays"] = dates.isin(values=holidays_instance.holidays)
mask_frame["holidays"] = dates.isin(values=holiday.holidays)
mask_frame["holiday_days_of_week"] = dates.dayofweek

mask_frame["business_days"] = dates.isin(
values=bizday_instance.get_business_days(dates=dates).index
)
mask_frame["business_days"] = bday.get_business_days(dates=dates)

for day, dow in zip(
["monday", "tuesday", "wednesday", "thursday", "friday"], range(5)
):
mask_frame[f"{day}_holidays"] = (
mask_frame["holiday_days_of_week"] == dow
) & mask_frame["holidays"]

mask_frame["eligible_days"] = (
mask_frame["business_days"] & ~mask_frame["holidays"]
)

return mask_frame
77 changes: 36 additions & 41 deletions fedcal/fedindex.py
Original file line number Diff line number Diff line change
Expand Up @@ -819,7 +819,7 @@ def posix_day(self) -> NDArray[int64]:
@property
def business_days(
self,
) -> NDArray[bool]:
) -> DatetimeIndex:
"""
Determine if the dates in the index are Federal business days.
Expand All @@ -828,16 +828,11 @@ def business_days(
Returns
-------
NDArray
An NDArray of boolean values, where True indicates a
business day.
Datetimeindex
Datetimeindex of dates that are business days.
"""
bizdays: FedBusDay = _date_attributes.FedBusDay()
next_business_days = self.datetimeindex + bizdays.get_business_days(
dates=self.datetimeindex
)
return self.datetimeindex.isin(values=next_business_days)
bdays: FedBusDay = _date_attributes.FedBusDay()
return self.datetimeindex[bdays.get_business_days(dates=self.datetimeindex)]

@property
def fys(self) -> Index[int]:
Expand Down Expand Up @@ -957,7 +952,7 @@ def fy_end(self) -> PeriodIndex:
return self._fiscalcal.fy_end

@property
def holidays(self) -> NDArray[bool]:
def holidays(self) -> DatetimeIndex:
"""
Identify federal holidays in the index.
Expand All @@ -966,15 +961,16 @@ def holidays(self) -> NDArray[bool]:
Returns
-------
NDArray
An NDArray of boolean values, where True indicates a federal
holiday.
DatetimeIndex
DatetimeIndex reflecting dates of holidays
"""
self._set_holidays()
return self.datetimeindex.isin(values=self._holidays.holidays)
return self.datetimeindex[
self.datetimeindex.isin(values=self._holidays.holidays)
]

@property
def proclaimed_holidays(self) -> NDArray[bool]:
def proclaimed_holidays(self) -> DatetimeIndex:
"""
Check for proclaimed federal holidays in the index.
Expand All @@ -983,39 +979,40 @@ def proclaimed_holidays(self) -> NDArray[bool]:
Returns
-------
NDArray
An NDArray of boolean values, where True indicates a proclaimed
federalholiday.
DatetimeIndex
DatetimeIndex reflecting dates of proclaimed holidays.
"""
self._set_holidays()
return self.datetimeindex.isin(
values=pd.DatetimeIndex(data=self._holidays.proclaimed_holidays)
)
return self.datetimeindex[
self.datetimeindex.isin(
values=pd.DatetimeIndex(data=self._holidays.proclaimed_holidays)
)
]

@property
def possible_proclamation_holidays(self) -> NDArray[bool]:
def possible_proclamation_holidays(self) -> DatetimeIndex:
"""
Guesses if the dates in the index are possible *future* proclamation
federal holidays.
Returns
-------
NDArray
NDArray of boolean values indicating possible proclamation
holidays.
DatetimeIndex
A DatetimeIndex reflecting possible *future* dates that could see a
proclaimed holiday.
Notes
-----
See notes to FedStamp.possible_proclamation_holiday.
"""
self._set_holidays()
return self._holidays.guess_proclamation_holidays(
datetimeindex=self.datetimeindex
)
return self.datetimeindex[
self._holidays.guess_proclamation_holidays(dates=self.datetimeindex)
]

@property
def probable_mil_passdays(self) -> NDArray[bool]:
def probable_mil_passdays(self) -> DatetimeIndex:
"""
Estimate military pass days within the index's date range.
Expand All @@ -1026,19 +1023,18 @@ def probable_mil_passdays(self) -> NDArray[bool]:
Returns
-------
NDArray
An array of boolean values, where True indicates a probable
military pass day.
DatetimeIndex
A datetimeindex reflecting probable dates for military passdays.
"""

passdays: ProbableMilitaryPassDay = _mil.ProbableMilitaryPassDay(
dates=self.datetimeindex
)
return passdays.passdays
return self.datetimeindex[passdays.passdays]

# Payday properties
@property
def mil_paydays(self) -> NDArray[bool]:
def mil_paydays(self) -> DatetimeIndex:
"""
Identify military payday dates within the index.
Expand All @@ -1047,11 +1043,11 @@ def mil_paydays(self) -> NDArray[bool]:
Returns
-------
NDArray
A boolean array indicating military payday dates as True.
DatetimeIndex
A datetimeindex reflecting military payday dates.
"""
milpays: MilitaryPayDay = _mil.MilitaryPayDay(dates=self.datetimeindex)
return milpays.paydays
return self.datetimeindex[milpays.paydays]

@property
def civ_paydays(self) -> Series[bool]:
Expand Down Expand Up @@ -1094,10 +1090,9 @@ def departments(self) -> DataFrame:
)

dept_df: DataFrame = self.datetimeindex.to_frame(name="Departments")
dhs_formed: Timestamp = time_utils.to_timestamp(constants.DHS_FORMED)
dept_df["Departments"] = dept_df.index.map(
mapper=lambda date: all_depts
if date >= constants.DHS_FORMED
else pre_dhs_depts
mapper=lambda date: all_depts if date >= dhs_formed else pre_dhs_depts
)
return dept_df

Expand Down

0 comments on commit 7cdf9e0

Please sign in to comment.