252 lines
7.7 KiB
Python
252 lines
7.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Centralized timezone utilities for the trading system
|
|
|
|
This module provides consistent timezone handling across all components:
|
|
- All external data (Binance, MEXC) comes in UTC
|
|
- All internal processing uses Europe/Sofia timezone
|
|
- All timestamps stored in database are timezone-aware
|
|
- All NN model inputs use consistent timezone
|
|
"""
|
|
|
|
import pytz
|
|
import pandas as pd
|
|
from datetime import datetime, timezone
|
|
from typing import Union, Optional
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Define timezone constants
|
|
UTC = pytz.UTC
|
|
SOFIA_TZ = pytz.timezone('Europe/Sofia')
|
|
SYSTEM_TIMEZONE = SOFIA_TZ # Our system's primary timezone
|
|
|
|
def get_system_timezone():
|
|
"""Get the system's primary timezone (Europe/Sofia)"""
|
|
return SYSTEM_TIMEZONE
|
|
|
|
def get_utc_timezone():
|
|
"""Get UTC timezone"""
|
|
return UTC
|
|
|
|
def now_utc() -> datetime:
|
|
"""Get current time in UTC"""
|
|
return datetime.now(UTC)
|
|
|
|
def now_sofia() -> datetime:
|
|
"""Get current time in Sofia timezone"""
|
|
return datetime.now(SOFIA_TZ)
|
|
|
|
def now_system() -> datetime:
|
|
"""Get current time in system timezone (Sofia)"""
|
|
return now_sofia()
|
|
|
|
def to_utc(dt: Union[datetime, pd.Timestamp]) -> datetime:
|
|
"""Convert datetime to UTC timezone"""
|
|
if dt is None:
|
|
return None
|
|
|
|
if isinstance(dt, pd.Timestamp):
|
|
dt = dt.to_pydatetime()
|
|
|
|
if dt.tzinfo is None:
|
|
# Assume it's in system timezone if no timezone info
|
|
dt = SYSTEM_TIMEZONE.localize(dt)
|
|
|
|
return dt.astimezone(UTC)
|
|
|
|
def to_sofia(dt: Union[datetime, pd.Timestamp]) -> datetime:
|
|
"""Convert datetime to Sofia timezone"""
|
|
if dt is None:
|
|
return None
|
|
|
|
if isinstance(dt, pd.Timestamp):
|
|
dt = dt.to_pydatetime()
|
|
|
|
if dt.tzinfo is None:
|
|
# Assume it's UTC if no timezone info (common for external data)
|
|
dt = UTC.localize(dt)
|
|
|
|
return dt.astimezone(SOFIA_TZ)
|
|
|
|
def to_system_timezone(dt: Union[datetime, pd.Timestamp]) -> datetime:
|
|
"""Convert datetime to system timezone (Sofia)"""
|
|
return to_sofia(dt)
|
|
|
|
def normalize_timestamp(timestamp: Union[int, float, str, datetime, pd.Timestamp],
|
|
source_tz: Optional[pytz.BaseTzInfo] = None) -> datetime:
|
|
"""
|
|
Normalize various timestamp formats to system timezone (Sofia)
|
|
|
|
Args:
|
|
timestamp: Timestamp in various formats
|
|
source_tz: Source timezone (defaults to UTC for external data)
|
|
|
|
Returns:
|
|
datetime: Timezone-aware datetime in system timezone
|
|
"""
|
|
if timestamp is None:
|
|
return now_system()
|
|
|
|
# Default source timezone is UTC (most external APIs use UTC)
|
|
if source_tz is None:
|
|
source_tz = UTC
|
|
|
|
try:
|
|
# Handle different timestamp formats
|
|
if isinstance(timestamp, (int, float)):
|
|
# Unix timestamp (assume seconds, convert to milliseconds if needed)
|
|
if timestamp > 1e10: # Milliseconds
|
|
timestamp = timestamp / 1000
|
|
dt = datetime.fromtimestamp(timestamp, tz=source_tz)
|
|
|
|
elif isinstance(timestamp, str):
|
|
# String timestamp
|
|
dt = pd.to_datetime(timestamp)
|
|
if dt.tzinfo is None:
|
|
dt = source_tz.localize(dt)
|
|
|
|
elif isinstance(timestamp, pd.Timestamp):
|
|
dt = timestamp.to_pydatetime()
|
|
if dt.tzinfo is None:
|
|
dt = source_tz.localize(dt)
|
|
|
|
elif isinstance(timestamp, datetime):
|
|
dt = timestamp
|
|
if dt.tzinfo is None:
|
|
dt = source_tz.localize(dt)
|
|
|
|
else:
|
|
logger.warning(f"Unknown timestamp format: {type(timestamp)}")
|
|
return now_system()
|
|
|
|
# Convert to system timezone
|
|
return dt.astimezone(SYSTEM_TIMEZONE)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error normalizing timestamp {timestamp}: {e}")
|
|
return now_system()
|
|
|
|
def normalize_dataframe_timestamps(df: pd.DataFrame,
|
|
timestamp_col: str = 'timestamp',
|
|
source_tz: Optional[pytz.BaseTzInfo] = None) -> pd.DataFrame:
|
|
"""
|
|
Normalize timestamps in a DataFrame to system timezone
|
|
|
|
Args:
|
|
df: DataFrame with timestamp column
|
|
timestamp_col: Name of timestamp column
|
|
source_tz: Source timezone (defaults to UTC)
|
|
|
|
Returns:
|
|
DataFrame with normalized timestamps
|
|
"""
|
|
if df.empty or timestamp_col not in df.columns:
|
|
return df
|
|
|
|
if source_tz is None:
|
|
source_tz = UTC
|
|
|
|
try:
|
|
# Convert to datetime if not already
|
|
if not pd.api.types.is_datetime64_any_dtype(df[timestamp_col]):
|
|
df[timestamp_col] = pd.to_datetime(df[timestamp_col])
|
|
|
|
# Handle timezone
|
|
if df[timestamp_col].dt.tz is None:
|
|
# Localize to source timezone first
|
|
df[timestamp_col] = df[timestamp_col].dt.tz_localize(source_tz)
|
|
|
|
# Convert to system timezone
|
|
df[timestamp_col] = df[timestamp_col].dt.tz_convert(SYSTEM_TIMEZONE)
|
|
|
|
return df
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error normalizing DataFrame timestamps: {e}")
|
|
return df
|
|
|
|
def normalize_dataframe_index(df: pd.DataFrame,
|
|
source_tz: Optional[pytz.BaseTzInfo] = None) -> pd.DataFrame:
|
|
"""
|
|
Normalize DataFrame index timestamps to system timezone
|
|
|
|
Args:
|
|
df: DataFrame with datetime index
|
|
source_tz: Source timezone (defaults to UTC)
|
|
|
|
Returns:
|
|
DataFrame with normalized index
|
|
"""
|
|
if df.empty or not isinstance(df.index, pd.DatetimeIndex):
|
|
return df
|
|
|
|
if source_tz is None:
|
|
source_tz = UTC
|
|
|
|
try:
|
|
# Handle timezone
|
|
if df.index.tz is None:
|
|
# Localize to source timezone first
|
|
df.index = df.index.tz_localize(source_tz)
|
|
|
|
# Convert to system timezone
|
|
df.index = df.index.tz_convert(SYSTEM_TIMEZONE)
|
|
|
|
return df
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error normalizing DataFrame index: {e}")
|
|
return df
|
|
|
|
def format_timestamp_for_display(dt: datetime, format_str: str = '%H:%M:%S') -> str:
|
|
"""
|
|
Format timestamp for display in system timezone
|
|
|
|
Args:
|
|
dt: Datetime to format
|
|
format_str: Format string
|
|
|
|
Returns:
|
|
Formatted timestamp string
|
|
"""
|
|
if dt is None:
|
|
return now_system().strftime(format_str)
|
|
|
|
try:
|
|
# Convert to system timezone if needed
|
|
if isinstance(dt, datetime):
|
|
if dt.tzinfo is None:
|
|
dt = UTC.localize(dt)
|
|
dt = dt.astimezone(SYSTEM_TIMEZONE)
|
|
|
|
return dt.strftime(format_str)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error formatting timestamp {dt}: {e}")
|
|
return now_system().strftime(format_str)
|
|
|
|
def get_timezone_offset_hours() -> float:
|
|
"""Get current timezone offset from UTC in hours"""
|
|
now = now_system()
|
|
utc_now = now_utc()
|
|
offset_seconds = (now - utc_now.replace(tzinfo=None)).total_seconds()
|
|
return offset_seconds / 3600
|
|
|
|
def is_market_hours() -> bool:
|
|
"""Check if it's currently market hours (24/7 for crypto, but useful for logging)"""
|
|
# Crypto markets are 24/7, but this can be useful for other purposes
|
|
return True
|
|
|
|
def log_timezone_info():
|
|
"""Log current timezone information for debugging"""
|
|
now_utc_time = now_utc()
|
|
now_sofia_time = now_sofia()
|
|
offset_hours = get_timezone_offset_hours()
|
|
|
|
logger.info(f"Timezone Info:")
|
|
logger.info(f" UTC Time: {now_utc_time}")
|
|
logger.info(f" Sofia Time: {now_sofia_time}")
|
|
logger.info(f" Offset: {offset_hours:+.1f} hours from UTC")
|
|
logger.info(f" System Timezone: {SYSTEM_TIMEZONE}") |