跨多个时间范围的开始日期以收集市场数据

Question

我正在编写一个交易算法，通过比较年度 OHLC、季度 OHLC、月度 OHLC、每周 OHLC 和每日 OHLC，来整合多个时间框架分析。

对于每个不同的时间范围，我需要前 4 个 OHLC 数据周期的开始日期，其中包括当前周期。

示例，相对于2022-01-09，发帖日期：

每年的时间范围 4 年开始日期：2019-01-02

季度时间框架 4 季度开始日期：2021-03-01

每月时间范围 4 个月开始日期：2021-10-01

每周时间范围 4 周开始日期：2021-12-13

每日时间范围 4 天开始日期：2022-01-04

是否有一个以金融市场为中心的 Python 模块，可以轻松引用这些类型的日期戳，我最终可以以上述示例的格式将其输出为字符串，以插入到我的多边形 REST 调用中？

我查看了各种 datetime 和 dateutil 类和方法，但它们都相当复杂，尽管它们似乎没有考虑到北美股票市场开放时的细微差别。

（如果存在可以理解英语表达的东西，那就太酷了；例如​​：“四年前一年中的第一个交易日”、“四个季度前本季度的第一个交易日”、“第一个交易日”一个月的交易日，四个月前'，'一周的第一个交易日，四个星期前'，'一周的一天，四个交易日之前'）。

解决方案

pandas 有一个非常全面的功能列表，可以满足我的需求。以下是我根据这些发现汇总的内容：

# timefunctions.py
import pandas
import time
import pandas_market_calendars as mcal

now = pandas.Timestamp.now()
tz = 'America/New York'
nyse = mcal.get_calendar('NYSE')

def start_of_year(multiplier, format='ms'):
    obj = now + pandas.tseries.offsets.BYearBegin(multiplier)
    obj = pandas.Timestamp(obj).floor(freq='D')
    if format == 'ms':
        obj = int(time.mktime(obj.timetuple()) * 1000)
    return(obj)

def start_of_quarter(multiplier, format='ms'):
    obj = now + pandas.tseries.offsets.BQuarterBegin(multiplier, startingMonth=1)
    obj = pandas.Timestamp(obj).floor(freq='D')
    if format == 'ms':
        obj = int(time.mktime(obj.timetuple()) * 1000)
    return(obj)

def start_of_month(multiplier, format='ms'):
    obj = now + pandas.tseries.offsets.BusinessMonthBegin(multiplier)
    obj = pandas.Timestamp(obj).floor(freq='D')
    if format == 'ms':
        obj = int(time.mktime(obj.timetuple()) * 1000)
    return(obj)

def start_of_week(multiplier, format='ms'):
    obj = now + pandas.tseries.offsets.Week(multiplier, weekday=0)
    obj = pandas.Timestamp(obj).floor(freq='D')
    if format == 'ms':
        obj = int(time.mktime(obj.timetuple()) * 1000)
    return(obj)

def start_of_day(multiplier, format='ms', action='signal'):
    """
    : action = 'signal' [default] (return timestamps of all trading days with market open times)
    : action = 'trigger' (return timestamps of today only) (ignores multiplier. hard coded to -1)
    : format = 'ms' [default] (return object in millisecond timestamps)
    """
    market_days = nyse.schedule(start_date=start_of_week(-1, None), end_date=now)
    if action == 'trigger':
        obj = market_days.index.to_list()[-1:][0]
        obj = obj + pandas.DateOffset(hour=nyse.open_time.hour, minute=nyse.open_time.minute, second=nyse.open_time.second)
        if format == 'ms':
            obj = int(time.mktime(obj.timetuple()) * 1000)
    elif action == 'signal':
        date_list = []
        date_list_ms = []
        for date_item in market_days.index.to_list()[multiplier:]:
            date_list.append(date_item + pandas.DateOffset(hour=nyse.open_time.hour, minute=nyse.open_time.minute, second=nyse.open_time.second))
        if format == 'ms':
            for date in date_list:
                date_list_ms.append(int(time.mktime(date.timetuple()) * 1000))
        if date_list:
            obj = date_list
        if date_list_ms:
            obj = date_list_ms
    return(obj)

输出：

In [1]: from modules import timefunctions

In [2]: timefunctions.start_of_year(-4, None)
Out[2]: Timestamp('2019-01-01 00:00:00')

In [3]: timefunctions.start_of_year(-4)
Out[3]: 1546318800000

In [4]: timefunctions.start_of_quarter(-4, None)
Out[4]: Timestamp('2021-04-01 00:00:00')

In [5]: timefunctions.start_of_quarter(-4)
Out[5]: 1617249600000

In [6]: timefunctions.start_of_month(-4, None)
Out[6]: Timestamp('2021-10-01 00:00:00')

In [7]: timefunctions.start_of_month(-4)
Out[7]: 1633060800000

In [8]: timefunctions.start_of_week(-4, None)
Out[8]: Timestamp('2021-12-27 00:00:00')

In [9]: timefunctions.start_of_week(-4)
Out[9]: 1640581200000

In [10]: timefunctions.start_of_day(-4, None)
Out[10]: 
[Timestamp('2022-01-18 09:30:00'),
 Timestamp('2022-01-19 09:30:00'),
 Timestamp('2022-01-20 09:30:00'),
 Timestamp('2022-01-21 09:30:00')]

In [11]: timefunctions.start_of_day(-4)
Out[11]: [1642516200000, 1642602600000, 1642689000000, 1642775400000]

In [12]: 

Answer 1

使用pandas 的解决方案更新了原始问题