从基础到高级详解Python时区处理的完全指南

import pytz
from datetime import datetime

# 创建时区对象
utc = pytz.utc
shanghai = pytz.timezone('Asia/Shanghai')
new_york = pytz.timezone('America/New_York')

# 时区关键属性
print(f"上海时区: {shanghai}")
print(f"时区名称: {shanghai.zone}")
print(f"UTC偏移: {shanghai.utcoffset(datetime.now())}")
print(f"夏令时规则: {shanghai.dst(datetime.now())}")

# 查看所有时区
all_timezones = pytz.all_timezones
print(f"总时区数: {len(all_timezones)}")
print("常见时区示例:")
for tz in ['UTC', 'Asia/Shanghai', 'Europe/London', 'America/New_York']:
    print(f"- {tz}")

# 创建原始时间
naive_dt = datetime(2023, 12, 15, 14, 30)

# 本地化（添加时区）
localized_sh = shanghai.localize(naive_dt)
print(f"上海时间: {localized_sh}")

# 转换为其他时区
ny_time = localized_sh.astimezone(new_york)
print(f"纽约时间: {ny_time}")

# UTC转换
utc_time = localized_sh.astimezone(utc)
print(f"UTC时间: {utc_time}")

def convert_time(source_time, source_tz, target_tz):
    """安全时区转换"""
    # 确保时区感知
    if source_time.tzinfo is None:
        source_time = source_tz.localize(source_time)
    
    return source_time.astimezone(target_tz)

# 使用示例
source = datetime(2023, 12, 15, 14, 30)
target = convert_time(source, shanghai, new_york)
print(f"转换结果: {source} [上海] -> {target} [纽约]")

# 创建跨越夏令时的日期
london = pytz.timezone('Europe/London')

# 夏令时开始前 (2023年3月25日)
pre_dst = datetime(2023, 3, 25, 12, 0)
pre_dst = london.localize(pre_dst)
print(f"夏令时前: {pre_dst} (UTC偏移: {pre_dst.utcoffset()})")

# 夏令时开始后 (2023年3月26日)
post_dst = datetime(2023, 3, 26, 12, 0)
post_dst = london.localize(post_dst)
print(f"夏令时后: {post_dst} (UTC偏移: {post_dst.utcoffset()})")

def handle_ambiguous_time(dt, tz):
    """处理模糊时间（夏令时结束）"""
    try:
        return tz.localize(dt, is_dst=None)  # 严格模式
    except pytz.AmbiguousTimeError:
        # 选择较早时间（标准时间）
        return tz.localize(dt, is_dst=False)
    except pytz.NonExistentTimeError:
        # 处理不存在时间（夏令时开始）
        return tz.localize(dt + timedelta(hours=1), is_dst=True)

# 测试模糊时间 (伦敦夏令时结束)
ambiguous_dt = datetime(2023, 10, 29, 1, 30)
try:
    # 直接尝试会报错
    london.localize(ambiguous_dt, is_dst=None)
except pytz.AmbiguousTimeError:
    print("检测到模糊时间")
    
# 安全处理
safe_dt = handle_ambiguous_time(ambiguous_dt, london)
print(f"处理结果: {safe_dt}")

# 方法1: 使用pytz
aware_dt = shanghai.localize(datetime(2023, 12, 15, 14, 30))

# 方法2: 使用datetime替换
aware_dt2 = datetime(2023, 12, 15, 14, 30, tzinfo=shanghai)

# 方法3: 使用ISO格式
aware_dt3 = datetime.fromisoformat("2023-12-15T14:30:00+08:00")

# 方法4: 使用dateutil
from dateutil import tz
aware_dt4 = datetime(2023, 12, 15, 14, 30, tzinfo=tz.gettz('Asia/Shanghai'))

# 时区感知比较
dt1 = shanghai.localize(datetime(2023, 12, 15, 14, 30))
dt2 = new_york.localize(datetime(2023, 12, 15, 1, 30))

print(f"时间相等: {dt1 == dt2}")  # False
print(f"时间顺序: {dt1 > dt2}")   # True

# 时区感知运算
duration = timedelta(hours=8)
new_time = dt1 + duration
print(f"8小时后: {new_time}")

# 跨时区运算
new_time_ny = new_time.astimezone(new_york)
print(f"纽约时间: {new_time_ny}")

class GlobalScheduler:
    """全球会议调度系统"""
    def __init__(self):
        self.user_timezones = {}
    
    def add_user(self, user_id, timezone):
        """添加用户时区"""
        self.user_timezones[user_id] = pytz.timezone(timezone)
    
    def find_meeting_time(self, users, duration_hours=1):
        """寻找共同可用时间"""
        # 获取所有用户时区
        timezones = [self.user_timezones[uid] for uid in users]
        
        # 找到重叠的工作时间 (示例算法)
        base_time = datetime.now(pytz.utc).replace(hour=0, minute=0, second=0)
        best_time = None
        max_overlap = 0
        
        # 检查未来7天
        for day in range(7):
            candidate = base_time + timedelta(days=day)
            
            # 检查每个小时段
            for hour in range(7, 19):  # 7AM-7PM
                start = candidate.replace(hour=hour)
                end = start + timedelta(hours=duration_hours)
                
                # 检查是否在所有时区都是工作时间
                valid = True
                for tz in timezones:
                    local_start = start.astimezone(tz)
                    local_end = end.astimezone(tz)
                    
                    # 检查是否在工作时间 (9AM-5PM)
                    if not (9 <= local_start.hour < 17 and 9 <= local_end.hour < 17):
                        valid = False
                        break
                
                if valid:
                    return start  # 找到第一个合适时间
        
        return None  # 未找到合适时间

# 使用示例
scheduler = GlobalScheduler()
scheduler.add_user('Alice', 'America/New_York')
scheduler.add_user('Bob', 'Europe/London')
scheduler.add_user('Charlie', 'Asia/Shanghai')

meeting_time = scheduler.find_meeting_time(['Alice', 'Bob', 'Charlie'])
print(f"会议时间 (UTC): {meeting_time}")

import schedule
import time

def timezone_aware_job():
    """时区感知任务"""
    now = datetime.now(pytz.utc)
    print(f"任务执行时间 (UTC): {now}")

# 创建调度器
scheduler = schedule.Scheduler()

# 纽约时间每天9:00执行
ny_tz = pytz.timezone('America/New_York')
ny_time = ny_tz.localize(datetime.now().replace(hour=9, minute=0, second=0))

# 转换为UTC
utc_time = ny_time.astimezone(pytz.utc)

# 添加任务
scheduler.every().day.at(utc_time.strftime('%H:%M')).do(timezone_aware_job)

# 运行调度器
while True:
    scheduler.run_pending()
    time.sleep(60)

class TradingHours:
    """全球交易时间处理器"""
    MARKET_HOURS = {
        'NYSE': ('America/New_York', (9, 30), (16, 0)),  # 9:30 AM - 4:00 PM
        'LSE': ('Europe/London', (8, 0), (16, 30)),      # 8:00 AM - 4:30 PM
        'TSE': ('Asia/Tokyo', (9, 0), (15, 0)),         # 9:00 AM - 3:00 PM
        'SHSE': ('Asia/Shanghai', (9, 30), (15, 0))      # 9:30 AM - 3:00 PM
    }
    
    def is_market_open(self, exchange, dt=None):
        """检查市场是否开放"""
        dt = dt or datetime.now(pytz.utc)
        tz_name, start_time, end_time = self.MARKET_HOURS[exchange]
        tz = pytz.timezone(tz_name)
        
        # 转换为市场时区
        market_time = dt.astimezone(tz)
        
        # 检查时间
        market_start = market_time.replace(hour=start_time[0], minute=start_time[1], second=0)
        market_end = market_time.replace(hour=end_time[0], minute=end_time[1], second=0)
        
        # 检查是否为工作日
        if market_time.weekday() >= 5:  # 周六日
            return False
        
        return market_start <= market_time <= market_end

# 使用示例
trader = TradingHours()
ny_time = datetime(2023, 12, 15, 14, 30, tzinfo=pytz.utc)  # UTC时间
print(f"纽交所开放: {trader.is_market_open('NYSE', ny_time)}")
print(f"上交所开放: {trader.is_market_open('SHSE', ny_time)}")

def convert_trade_time(trade_time, source_exchange, target_exchange):
    """转换交易时间到另一交易所时区"""
    trader = TradingHours()
    source_tz = pytz.timezone(trader.MARKET_HOURS[source_exchange][0])
    target_tz = pytz.timezone(trader.MARKET_HOURS[target_exchange][0])
    
    # 确保时区感知
    if trade_time.tzinfo is None:
        trade_time = source_tz.localize(trade_time)
    
    return trade_time.astimezone(target_tz)

# 使用示例
ny_trade = datetime(2023, 12, 15, 10, 30)  # 纽约时间10:30 AM
ny_trade = pytz.timezone('America/New_York').localize(ny_trade)

sh_trade = convert_trade_time(ny_trade, 'NYSE', 'SHSE')
print(f"纽约交易时间: {ny_trade}")
print(f"上海对应时间: {sh_trade}")

def normalize_timezone(df, time_col, target_tz='UTC'):
    """标准化DataFrame时区"""
    # 转换为时区感知
    if df[time_col].dt.tz is None:
        # 假设为UTC（实际应用中需根据数据源确定）
        df[time_col] = df[time_col].dt.tz_localize('UTC')
    
    # 转换为目标时区
    df[time_col] = df[time_col].dt.tz_convert(target_tz)
    return df

# 使用示例
import pandas as pd

# 创建多时区数据
data = {
    'timestamp': [
        datetime(2023, 12, 15, 9, 30, tzinfo=pytz.timezone('America/New_York')),
        datetime(2023, 12, 15, 14, 30, tzinfo=pytz.timezone('Europe/London')),
        datetime(2023, 12, 15, 22, 30, tzinfo=pytz.timezone('Asia/Shanghai'))
    ],
    'value': [100, 200, 300]
}
df = pd.DataFrame(data)

# 标准化为UTC
df_normalized = normalize_timezone(df, 'timestamp', 'UTC')
print("标准化后数据:")
print(df_normalized)

def analyze_by_timezone(df, time_col, value_col):
    """按原始时区分组分析"""
    # 提取原始时区
    df['original_tz'] = df[time_col].apply(lambda x: x.tzinfo.zone)
    
    # 转换为本地时间
    df['local_time'] = df[time_col].apply(lambda x: x.tz_convert(x.tzinfo).time())
    
    # 按时区和小时分组
    result = df.groupby(['original_tz', df['local_time'].apply(lambda x: x.hour)])[value_col].mean()
    return result

# 使用示例
result = analyze_by_timezone(df, 'timestamp', 'value')
print("按时区分组分析:")
print(result)

import psycopg2
from psycopg2 import sql

def setup_database_timezone():
    """配置数据库时区"""
    conn = psycopg2.connect(dbname='test', user='postgres', password='password')
    cur = conn.cursor()
    
    # 设置数据库时区为UTC
    cur.execute("SET TIME ZONE 'UTC';")
    
    # 创建带时区的时间戳字段
    cur.execute("""
        CREATE TABLE events (
            id SERIAL PRIMARY KEY,
            name VARCHAR(100),
            event_time TIMESTAMPTZ NOT NULL
        )
    """)
    
    # 插入时区感知时间
    event_time = datetime.now(pytz.utc)
    cur.execute(
        sql.SQL("INSERT INTO events (name, event_time) VALUES (%s, %s)"),
        ('Meeting', event_time)
    )
    
    conn.commit()
    cur.close()
    conn.close()

# 从数据库读取
def read_events_in_timezone(target_tz):
    """在特定时区读取事件"""
    conn = psycopg2.connect(dbname='test', user='postgres', password='password')
    cur = conn.cursor()
    
    # 转换时区
    cur.execute("""
        SELECT name, event_time AT TIME ZONE %s AS local_time
        FROM events
    """, (target_tz,))
    
    results = cur.fetchall()
    cur.close()
    conn.close()
    return results

# 使用示例
setup_database_timezone()
events = read_events_in_timezone('Asia/Shanghai')
print("上海时间事件:")
for name, time in events:
    print(f"{name}: {time}")

from pymongo import MongoClient
import pytz

def store_datetime_with_timezone():
    """存储带时区的时间"""
    client = MongoClient('localhost', 27017)
    db = client['test_db']
    collection = db['events']
    
    # 创建时区感知时间
    event_time = datetime.now(pytz.utc)
    
    # 插入文档
    collection.insert_one({
        'name': 'Product Launch',
        'time': event_time
    })
    
    # 查询并转换时区
    shanghai_tz = pytz.timezone('Asia/Shanghai')
    for doc in collection.find():
        # 转换时区
        utc_time = doc['time']
        sh_time = utc_time.astimezone(shanghai_tz)
        print(f"事件时间 (上海): {sh_time}")

# 使用示例
store_datetime_with_timezone()

# 始终以UTC存储时间
def save_event(event_time):
    if event_time.tzinfo is None:
        event_time = pytz.utc.localize(event_time)
    else:
        event_time = event_time.astimezone(pytz.utc)
    db.save(event_time)

def display_time(event_time, user_tz):
    """根据用户时区显示时间"""
    return event_time.astimezone(user_tz).strftime('%Y-%m-%d %H:%M %Z')

# 错误做法
dt = datetime(2023, 12, 15, tzinfo=pytz.timezone('Asia/Shanghai'))

# 正确做法
dt = pytz.timezone('Asia/Shanghai').localize(datetime(2023, 12, 15))

def handle_dst_transition(dt, tz):
    try:
        return tz.localize(dt, is_dst=None)
    except (pytz.AmbiguousTimeError, pytz.NonExistentTimeError):
        # 使用业务逻辑处理
        return tz.localize(dt, is_dst=False)

def safe_time_compare(dt1, dt2):
    """安全比较时间"""
    # 确保时区感知
    if dt1.tzinfo is None:
        dt1 = pytz.utc.localize(dt1)
    if dt2.tzinfo is None:
        dt2 = pytz.utc.localize(dt2)

    # 转换为UTC比较
    return dt1.astimezone(pytz.utc) < dt2.astimezone(pytz.utc)

class GlobalEventSystem:
    """全球事件排序系统"""
    def __init__(self):
        self.events = []
    
    def add_event(self, event, location):
        """添加事件（带位置）"""
        tz = pytz.timezone(self._location_to_tz(location))
        if not event['time'].tzinfo:
            event['time'] = tz.localize(event['time'])
        self.events.append(event)
    
    def get_ordered_events(self):
        """获取全局排序事件"""
        # 转换为UTC排序
        utc_events = [{
            'event': e,
            'utc_time': e['time'].astimezone(pytz.utc)
        } for e in self.events]
        
        # 排序
        sorted_events = sorted(utc_events, key=lambda x: x['utc_time'])
        return [e['event'] for e in sorted_events]
    
    def _location_to_tz(self, location):
        """位置转时区（简化版）"""
        mapping = {
            'New York': 'America/New_York',
            'London': 'Europe/London',
            'Shanghai': 'Asia/Shanghai',
            'Tokyo': 'Asia/Tokyo'
        }
        return mapping.get(location, 'UTC')

# 使用示例
system = GlobalEventSystem()

# 添加全球事件
system.add_event({'name': '会议', 'time': datetime(2023, 12, 15, 9, 30)}, 'New York')
system.add_event({'name': '发布', 'time': datetime(2023, 12, 15, 14, 30)}, 'London')
system.add_event({'name': '上线', 'time': datetime(2023, 12, 15, 22, 30)}, 'Shanghai')

# 获取全局排序
ordered = system.get_ordered_events()
print("全局事件顺序:")
for event in ordered:
    print(f"{event['name']}: {event['time']}")

class TimezoneAwareLogger:
    """时区感知日志系统"""
    def __init__(self, default_tz='UTC'):
        self.default_tz = pytz.timezone(default_tz)
        self.logs = []
    
    def log(self, message, timestamp=None, timezone=None):
        """记录日志"""
        if timestamp is None:
            timestamp = datetime.now(pytz.utc)
        elif timestamp.tzinfo is None:
            tz = pytz.timezone(timezone) if timezone else self.default_tz
            timestamp = tz.localize(timestamp)
        
        self.logs.append({
            'message': message,
            'timestamp': timestamp,
            'timezone': timestamp.tzinfo.zone
        })
    
    def display_logs(self, target_tz='UTC'):
        """在特定时区显示日志"""
        target_tz = pytz.timezone(target_tz)
        for log in self.logs:
            local_time = log['timestamp'].astimezone(target_tz)
            print(f"[{local_time}] {log['message']}")

# 使用示例
logger = TimezoneAwareLogger()

# 记录日志（不同时区）
logger.log("系统启动", timezone='Asia/Shanghai')
logger.log("用户登录", datetime(2023, 12, 15, 9, 30), 'America/New_York')
logger.log("错误发生", timezone='Europe/London')

# 在UTC查看
print("UTC日志:")
logger.display_logs('UTC')

# 在纽约查看
print("\n纽约日志:")
logger.display_logs('America/New_York')

def process_event_time(event_time, event_tz):
    """
    处理事件时间

    参数:
        event_time: 事件时间 (datetime对象)
        event_tz: 事件时区 (字符串或tzinfo)

    返回:
        标准化UTC时间

    注意:
        如果event_time未指定时区，使用event_tz本地化
    """
    # 实现代码