Skip to content

broker.calendar.nyse_calendar

broker.calendar.nyse_calendar

NYSECalendar 

NYSECalendar()

Bases: Calendar

Trading calendar for the New York Stock Exchange (9:00AM-4:00PM on weekdays, holidays taken into account).

Initializes `NYSECalendar.

Source code in src\contango\broker\calendar\nyse_calendar.py 
18
19
20
21
22
def __init__(self) -> None:
    """
    Initializes `NYSECalendar.
    """
    self._calendar = xcals.get_calendar("XNYS", start="1990-01-01")

get_expected_timestamps 

get_expected_timestamps(start_timestamp: int, end_timestamp: int, interval: Interval) -> list[datetime]

Returns the expected available NYSE timestamps for the specified period.

Parameters:

Name Type Description Default
start_timestamp int

The start time in unix ms.

 required
end_timestamp int

The end time in unix ms.

 required
interval Interval

The bar interval type.

 required

Returns:

Type Description
list[datetime]

list[datetime]: The close times of the available NYSE trading bars for the designated period.
Source code in src\contango\broker\calendar\nyse_calendar.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def get_expected_timestamps(
    self,
    start_timestamp: int,
    end_timestamp: int,
    interval: Interval,
) -> list[datetime]:
    """
    Returns the expected available NYSE timestamps for the specified period.

    Args:
        start_timestamp: The start time in unix ms.
        end_timestamp: The end time in unix ms.
        interval: The bar interval type.

    Returns:
        list[datetime]: The close times of the available NYSE trading bars for the designated period.
    """
    start_ny = pd.Timestamp(start_timestamp, unit="ms", tz=ZoneInfo("UTC")).tz_convert(ZoneInfo("America/New_York"))
    end_ny = pd.Timestamp(end_timestamp, unit="ms", tz=ZoneInfo("UTC")).tz_convert(ZoneInfo("America/New_York"))

    start_date = start_ny.tz_localize(None).normalize()
    end_date = end_ny.tz_localize(None).normalize()

    frequency = pd.Timedelta(interval.value)

    raw_index = self._calendar.trading_index(
        start=start_date,
        end=end_date,
        period=frequency,
        intervals=True,
        force=True,
    )

    close_timestamps: pd.DatetimeIndex
    if isinstance(raw_index, pd.IntervalIndex):
        close_timestamps = cast(pd.DatetimeIndex, raw_index.right)
    else:
        close_timestamps = raw_index.tz_localize(timezone.utc)

    start_utc = start_ny.tz_convert(ZoneInfo("UTC"))
    end_utc = end_ny.tz_convert(ZoneInfo("UTC"))

    close_times: list[datetime] = [
        ts.to_pydatetime().astimezone(timezone.utc)
        for ts in close_timestamps
        if start_utc <= ts <= end_utc
    ]

    return close_times