Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add the calendars module #2013

Merged
merged 5 commits into from
Mar 30, 2022
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
372 changes: 372 additions & 0 deletions pyintegration/deephaven2/calendars.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,372 @@
#
jmao-denver marked this conversation as resolved.
Show resolved Hide resolved
# Copyright (c) 2016-2022 Deephaven Data Labs and Patent Pending
#
""" This module provides access to a collection of business calendars and convenient calendar functions. """
jmao-denver marked this conversation as resolved.
Show resolved Hide resolved

from enum import Enum
from typing import List

import jpy
from deephaven2.time import TimeZone, DateTime

from deephaven2 import DHError
from deephaven2._wrapper_abc import JObjectWrapper

_JCalendars = jpy.get_type("io.deephaven.time.calendar.Calendars")
_JCalendar = jpy.get_type("io.deephaven.time.calendar.Calendar")
_JDayOfWeek = jpy.get_type("java.time.DayOfWeek")
_JBusinessPeriod = jpy.get_type("io.deephaven.time.calendar.BusinessPeriod")
_JBusinessSchedule = jpy.get_type("io.deephaven.time.calendar.BusinessSchedule")
_JBusinessCalendar = jpy.get_type("io.deephaven.time.calendar.BusinessCalendar")


def calendar_names() -> List[str]:
""" Returns the names of all available calendars.

Returns:
a list of names of all available calendars

Raises:
DHError
"""
try:
return list(_JCalendars.calendarNames())
except Exception as e:
raise DHError(e, "failed to obtain the available calendar names.") from e


def default_calendar_name():
""" Returns the default calendar name which is set by the 'Calendar.default' property in the configuration file
that the Deephaven server is started with.

Returns:
the default business calendar name

Raises:
DHError
"""
try:
return _JCalendars.getDefaultName()
except Exception as e:
raise DHError(e, "failed to get the default calendar name.") from e


class DayOfWeek(Enum):
""" A Enum that defines the days of a week. """
MONDAY = _JDayOfWeek.MONDAY
TUESDAY = _JDayOfWeek.TUESDAY
WEDNESDAY = _JDayOfWeek.WEDNESDAY
THURSDAY = _JDayOfWeek.THURSDAY
FRIDAY = _JDayOfWeek.FRIDAY
SATURDAY = _JDayOfWeek.SATURDAY
SUNDAY = _JDayOfWeek.SUNDAY
jmao-denver marked this conversation as resolved.
Show resolved Hide resolved


class BusinessPeriod(JObjectWrapper):
""" A period of business time during a business day. """

j_object_type = _JBusinessPeriod

def __init__(self, j_business_period):
self.j_business_period = j_business_period

def __repr__(self):
return f"[{self.start_time}, {self.end_time}]"

@property
def j_object(self) -> jpy.JType:
return self.j_business_period

@property
def start_time(self) -> DateTime:
""" The start of the period. """
return self.j_business_period.getStartTime()

@property
def end_time(self) -> DateTime:
""" The end of the period. """
return self.j_business_period.getEndTime()

@property
def length(self) -> int:
""" The length of the period in nanoseconds. """
return self.j_business_period.getLength()


class BusinessSchedule(JObjectWrapper):
""" A business schedule defines a single business day. """
j_object_type = _JBusinessSchedule

def __init__(self, j_business_schedule):
self.j_business_schedule = j_business_schedule

@property
def j_object(self) -> jpy.JType:
return self.j_business_schedule

@property
def business_periods(self) -> List[BusinessPeriod]:
""" The business periods of the day.

Returns:
List[BusinessPeriod]
"""
j_periods = self.j_business_schedule.getBusinessPeriods()
periods = []
for j_period in j_periods:
periods.append(BusinessPeriod(j_period))
return periods

@property
def start_of_day(self) -> DateTime:
""" The start of the business day. """
return self.j_business_schedule.getStartOfBusinessDay()

@property
def end_of_day(self) -> DateTime:
""" The end of the business day. """
return self.j_business_schedule.getEndOfBusinessDay()

def is_business_day(self) -> bool:
""" Whether it is a business day.

Returns:
bool
"""
return self.j_business_schedule.isBusinessDay()

def is_business_time(self, time: DateTime) -> bool:
""" Whether the specified time is a business time for the day.

Args:
time (DateTime): the time during the day

Return:
bool
"""
return self.j_business_schedule.isBusinessTime(time)

def business_time_elapsed(self, time: DateTime) -> int:
""" Returns the amount of business time in nanoseconds that has elapsed on the given day by the specified
time.

Args:
time (DateTime): the time during the day

Returns:
int
"""
return self.j_business_schedule.businessTimeElapsed(time)


class BusinessCalendar(JObjectWrapper):
jmao-denver marked this conversation as resolved.
Show resolved Hide resolved
""" A business calendar. """

j_object_type = _JBusinessCalendar

def __init__(self, name: str = None):
""" Creates a business calendar.

Args:
name (str) : name of the calendar, default is None, which means to use the default Deephaven calendar
jmao-denver marked this conversation as resolved.
Show resolved Hide resolved

Raises:
DHError
"""
try:
if name:
self.name = name
self.j_calendar = _JCalendars.calendar(name)
else:
self.name = default_calendar_name()
self.j_calendar = _JCalendars.calendar()
except Exception as e:
raise DHError(e, "failed to create a business calendar.") from e

@property
def j_object(self) -> jpy.JType:
return self.j_calendar

@property
def current_day(self) -> str:
""" The current day. """
return self.j_calendar.currentDay()

@property
def previous_day(self) -> str:
""" The previous day. """
return self.j_calendar.previousDay()

@property
def next_day(self) -> str:
""" The next day. """
return self.j_calendar.nextDay()

@property
def day_of_week(self) -> DayOfWeek:
""" The day of week for today. """
return DayOfWeek(self.j_calendar.dayOfWeek())

@property
def time_zone(self) -> TimeZone:
""" Returns the timezone of th e calendar. """
jmao-denver marked this conversation as resolved.
Show resolved Hide resolved
return TimeZone(self.j_calendar.timeZone())

@property
def is_business_day(self) -> bool:
""" If today is a business day. """
return self.j_calendar.isBusinessDay()

@property
def default_business_periods(self) -> List[str]:
""" The default business periods for the business days. Returns a list of strings with a comma separating
open and close times. """
default_business_periods = self.j_calendar.getDefaultBusinessPeriods()
b_periods = []
for i in range(default_business_periods.size()):
p = default_business_periods.get(i)
b_periods.append(p)
jmao-denver marked this conversation as resolved.
Show resolved Hide resolved

return b_periods

@property
def standard_business_day_length(self) -> int:
""" The length of a standard business day in nanoseconds. """
return self.j_calendar.standardBusinessDayLengthNanos()

def get_business_schedule(self, date: str) -> BusinessSchedule:
jmao-denver marked this conversation as resolved.
Show resolved Hide resolved
""" Returns the specified day's business schedule.

Args:
date (str): the date str, format must be "yyyy-MM-dd"

Returns:
a BusinessSchedule instance
"""
return BusinessSchedule(j_business_schedule=self.j_calendar.getBusinessSchedule(date))

def previous_business_day(self, date: str) -> str:
""" Gets the previous business day of the given date.
jmao-denver marked this conversation as resolved.
Show resolved Hide resolved

Args:
date (str): the date of interest

Returns:
str
"""
return self.j_calendar.previousBusinessDay(date)

def previous_non_business_day(self, date: str) -> str:
""" Gets the previous non-business day of the given date.
jmao-denver marked this conversation as resolved.
Show resolved Hide resolved

Args:
date (str): the date of interest

Returns:
str
"""
return self.j_calendar.previousNonBusinessDay(date)

def next_business_day(self, date: str) -> str:
""" Gets the next business day of the given date.
jmao-denver marked this conversation as resolved.
Show resolved Hide resolved

Args:
date (str): the date of interest

Returns:
str
"""
return self.j_calendar.nextBusinessDay(date)

def next_non_business_day(self, date: str) -> str:
""" Gets the next non-business day of the given date.
jmao-denver marked this conversation as resolved.
Show resolved Hide resolved

Args:
date (str): the date of interest

Returns:
str
"""
return self.j_calendar.nextNonBusinessDay(date)

def business_days_in_range(self, start: str, end: str) -> List[str]:
""" Returns the business days between the specified start and end dates.

Args:
start (str): the start day of the range
end (str): the end day of the range

Returns:
List[str]: a list of dates
"""
j_days = self.j_calendar.businessDaysInRange(start, end)
days = []
for day in j_days:
days.append(day)

return days

def non_business_days_in_range(self, start: str, end: str) -> List[str]:
""" Returns the non-business days between the specified start and end dates.

Args:
start (str): the start day of the range
end (str): the end day of the range

Returns:
List[str]: a list of dates
"""
j_days = self.j_calendar.nonBusinessDaysInRange(start, end)
days = []
for day in j_days:
days.append(day)

return days

def number_of_business_days(self, start: str, end: str, end_inclusive: bool = False) -> int:
""" Returns the number of business days between the start and end dates.

Args:
start (str): the start day of the range
end (str): the end day of the range
end_inclusive (bool): whether to include the end date, default is False

Returns:
int
"""
return self.j_calendar.numberOfBusinessDays(start, end, end_inclusive)

def number_of_non_business_days(self, start: str, end: str, end_inclusive: bool = False) -> int:
""" Returns the number of non-business days between the start and end dates.

Args:
start (str): the start day of the range
end (str): the end day of the range
end_inclusive (bool): whether to include the end date, default is False

Returns:
int
"""
return self.j_calendar.numberOfNonBusinessDays(start, end, end_inclusive)

def is_last_business_day_of_month(self, date: str):
""" Returns if the specified date is the last business day of the month.

Args:
date (str): the date

Returns:
bool
"""
return self.j_calendar.isLastBusinessDayOfMonth(date)

def is_last_business_day_of_week(self, date: str):
""" Returns if the specified date is the last business day of the week.

Args:
date (str): the date

Returns:
bool
"""
return self.j_calendar.isLastBusinessDayOfWeek(date)
4 changes: 2 additions & 2 deletions pyintegration/deephaven2/time.py
Original file line number Diff line number Diff line change
Expand Up @@ -162,8 +162,8 @@ def datetime_at_midnight(dt: DateTime, tz: TimeZone) -> DateTime:
""" Returns a DateTime for the requested DateTime at midnight in the specified time zone.

Args:
dt (DateTime) - DateTime for which the new value at midnight should be calculated
tz: (TimeZone) - TimeZone for which the new value at midnight should be calculated
dt (DateTime): the DateTime for which the new value at midnight should be calculated
tz (TimeZone): the TimeZone to use when interpreting the DateTime

Returns:
DateTime
Expand Down
Loading