a836cda1b2
Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import date, datetime, time, timedelta, timezone
|
|
from typing import Any
|
|
from zoneinfo import ZoneInfo
|
|
|
|
SHANGHAI = ZoneInfo("Asia/Shanghai")
|
|
|
|
|
|
def now_shanghai(clock: datetime | None = None) -> datetime:
|
|
if clock is not None:
|
|
if clock.tzinfo is None:
|
|
return clock.replace(tzinfo=SHANGHAI)
|
|
return clock.astimezone(SHANGHAI)
|
|
return datetime.now(SHANGHAI)
|
|
|
|
|
|
def isoformat(value: datetime | None = None) -> str:
|
|
current = now_shanghai(value)
|
|
return current.isoformat(timespec="seconds")
|
|
|
|
|
|
def yyyymmdd(value: date | datetime | str | None = None) -> str:
|
|
if value is None:
|
|
return now_shanghai().strftime("%Y%m%d")
|
|
if isinstance(value, str):
|
|
digits = value.replace("-", "")[:8]
|
|
if len(digits) != 8 or not digits.isdigit():
|
|
raise ValueError(f"invalid trade_date: {value}")
|
|
return digits
|
|
if isinstance(value, datetime):
|
|
return value.astimezone(SHANGHAI).strftime("%Y%m%d")
|
|
return value.strftime("%Y%m%d")
|
|
|
|
|
|
def parse_trade_date(value: str) -> date:
|
|
text = yyyymmdd(value)
|
|
return date(int(text[:4]), int(text[4:6]), int(text[6:8]))
|
|
|
|
|
|
def session_phase(clock: datetime | None, is_open_day: bool) -> str:
|
|
"""pre | intradaily | lunch | eod | closed"""
|
|
if not is_open_day:
|
|
return "closed"
|
|
current = now_shanghai(clock).time()
|
|
if current < time(9, 15):
|
|
return "pre"
|
|
if current < time(11, 30) or (time(13, 0) <= current <= time(15, 5)):
|
|
return "intraday"
|
|
if current < time(13, 0):
|
|
return "lunch"
|
|
if current <= time(23, 40):
|
|
return "eod"
|
|
return "closed"
|
|
|
|
|
|
def add_days(trade_date: str, days: int) -> str:
|
|
return (parse_trade_date(trade_date) + timedelta(days=days)).strftime("%Y%m%d")
|
|
|
|
|
|
def iter_yyyymmdd(start: str, end: str):
|
|
cursor = parse_trade_date(start)
|
|
last = parse_trade_date(end)
|
|
if cursor > last:
|
|
return
|
|
while cursor <= last:
|
|
yield cursor.strftime("%Y%m%d")
|
|
cursor += timedelta(days=1)
|
|
|
|
|
|
def utc_timestamp(value: Any) -> str:
|
|
if isinstance(value, datetime):
|
|
return isoformat(value)
|
|
return isoformat()
|