123 lines
3.6 KiB
Python
123 lines
3.6 KiB
Python
"""
|
|
Utility functions for dexorder.
|
|
|
|
Includes timestamp conversions, date parsing, and other common utilities.
|
|
All internal timestamps use nanoseconds since epoch (UTC).
|
|
"""
|
|
|
|
import logging
|
|
import re
|
|
from typing import Union
|
|
from datetime import datetime, timezone
|
|
import pandas as pd
|
|
from dateutil import parser as _dateutil_parser
|
|
from dateutil.relativedelta import relativedelta
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# Type alias for flexible timestamp input
|
|
TimestampInput = Union[int, float, str, datetime, pd.Timestamp]
|
|
|
|
NANOS_PER_SECOND = 1_000_000_000
|
|
|
|
_RELATIVE_RE = re.compile(
|
|
r'^(\d+)\s+(second|minute|hour|day|week|month|year)s?\s+ago$',
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def _parse_relative_date(s: str) -> datetime | None:
|
|
"""Parse relative date strings like '30 days ago', '2 weeks ago'."""
|
|
m = _RELATIVE_RE.match(s.strip())
|
|
if not m:
|
|
return None
|
|
n, unit = int(m.group(1)), m.group(2).lower()
|
|
return datetime.now(timezone.utc) - relativedelta(**{f'{unit}s': n})
|
|
|
|
|
|
def to_nanoseconds(timestamp: TimestampInput) -> int:
|
|
"""
|
|
Convert various timestamp formats to nanoseconds since epoch.
|
|
|
|
This is the canonical way to convert user-friendly timestamps (unix seconds,
|
|
date strings, datetime objects) into the internal nanosecond format used
|
|
throughout the dexorder system.
|
|
|
|
Args:
|
|
timestamp: Can be:
|
|
- Unix timestamp (int/float) - assumed to be in seconds
|
|
- ISO date string (str) - parsed using dateutil
|
|
- Relative date string (str) - e.g. "30 days ago", "2 weeks ago"
|
|
- datetime object
|
|
- pandas Timestamp
|
|
|
|
Returns:
|
|
Nanoseconds since epoch as integer
|
|
|
|
Examples:
|
|
>>> to_nanoseconds(1640000000) # Unix timestamp in seconds
|
|
1640000000000000000
|
|
>>> to_nanoseconds(1640000000.5) # Unix timestamp with fractional seconds
|
|
1640000000500000000
|
|
>>> to_nanoseconds("2021-12-20")
|
|
1639958400000000000
|
|
"""
|
|
if isinstance(timestamp, (int, float)):
|
|
return int(timestamp * NANOS_PER_SECOND)
|
|
elif isinstance(timestamp, str):
|
|
dt = _parse_relative_date(timestamp)
|
|
if dt is None:
|
|
dt = _dateutil_parser.parse(timestamp)
|
|
if dt is None:
|
|
raise ValueError(f"Could not parse date string: {timestamp}")
|
|
return int(dt.timestamp() * NANOS_PER_SECOND)
|
|
elif isinstance(timestamp, datetime):
|
|
return int(timestamp.timestamp() * NANOS_PER_SECOND)
|
|
elif isinstance(timestamp, pd.Timestamp):
|
|
return int(timestamp.timestamp() * NANOS_PER_SECOND)
|
|
else:
|
|
raise TypeError(f"Unsupported timestamp type: {type(timestamp)}")
|
|
|
|
|
|
def to_seconds(timestamp_nanos: int) -> float:
|
|
"""
|
|
Convert nanoseconds since epoch to Unix timestamp in seconds.
|
|
|
|
Args:
|
|
timestamp_nanos: Timestamp in nanoseconds since epoch
|
|
|
|
Returns:
|
|
Unix timestamp in seconds (float)
|
|
|
|
Examples:
|
|
>>> to_seconds(1640000000000000000)
|
|
1640000000.0
|
|
"""
|
|
return timestamp_nanos / NANOS_PER_SECOND
|
|
|
|
|
|
def to_datetime(timestamp_nanos: int) -> datetime:
|
|
"""
|
|
Convert nanoseconds since epoch to datetime object (UTC).
|
|
|
|
Args:
|
|
timestamp_nanos: Timestamp in nanoseconds since epoch
|
|
|
|
Returns:
|
|
datetime object in UTC
|
|
"""
|
|
return datetime.fromtimestamp(timestamp_nanos / NANOS_PER_SECOND)
|
|
|
|
|
|
def to_timestamp(timestamp_nanos: int) -> pd.Timestamp:
|
|
"""
|
|
Convert nanoseconds since epoch to pandas Timestamp.
|
|
|
|
Args:
|
|
timestamp_nanos: Timestamp in nanoseconds since epoch
|
|
|
|
Returns:
|
|
pandas Timestamp
|
|
"""
|
|
return pd.Timestamp(timestamp_nanos, unit='ns')
|