如何正确将yfinance带时区Timestamp写入InfluxDB并查询？

问题场景 #

通过yfinance获取的股票数据时间戳包含America/New_York时区信息：

ticker = 'AAPL'
import yfinance as yf
df = yf.Ticker('AAPL').history(period="1d").index[0]
print(df)

输出：

Timestamp('2023-01-05 00:00:00-0500', tz='America/New_York')

但写入InfluxDB后查询得到的时间自动转换为UTC时区：

df['_time']

输出：

0   2023-01-05 05:00:00+00:00
Name: _time, dtype: datetime64[ns, tzutc()]

完整写入/查询代码 #

写入代码 #

import yfinance as yf
import influxdb_client
from influxdb_client.client.write_api import SYNCHRONOUS, PointSettings


token = "my-token"
org = "my-org"
url = "my-url"
bucket = "stocks_us"
retention_policy = "autogen"

client = influxdb_client.InfluxDBClient(url=url, token=token, org=org)
write_api = client.write_api(write_options=SYNCHRONOUS)

df = yf.Ticker('AAPL').history(period="1d")

with client:
    """
    Ingest DataFrame with default tags
    """
    point_settings = PointSettings(**{"ticker": ticker})
    
    write_api = client.write_api(write_options=SYNCHRONOUS, 
                                 point_settings=point_settings)
    write_api.write(bucket=bucket, 
                    org= "dev", 
                    record=df, 
                    data_frame_measurement_name="stock_daily_df")

    client.close()

    print(df)

查询代码 #

import influxdb_client 

token = "my-token"
org = "my-org"
url = "my-url"
bucket = "stocks_us"
retention_policy = "autogen"

client = influxdb_client.InfluxDBClient(url=url, token=token, org=org)

query_api = client.query_api()
measurement= "stock_daily_df"

with client:
    
    """
    Querying ingested data
    """
    query = 'from(bucket:"{}")' \
            ' |> range(start: 0, stop: now())' \
            ' |> filter(fn: (r) => r._measurement == "{}")' \
            ' |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")' \
            ' |> filter(fn: (r) => r["ticker"] == "AAPL")'\
            ' |> limit(n:10, offset: 0)'.format(bucket, measurement)
            
    df = query_api.query_data_frame(query=query)

    print(df)

解决方案 #

原理说明 #

InfluxDB默认以UTC时间存储所有时间戳，这是行业通用规范，可避免跨时区数据混乱。以下两种方式可确保时间展示与原始时区一致：

1. 查询时将UTC时间转换回原始时区（推荐）

修改Flux查询语句，通过timeZone参数或convertTimezone函数，将UTC时间转换为America/New_York时区：

# 方式1：在range阶段指定时区
query = 'from(bucket:"{}")' \
        ' |> range(start: 0, stop: now(), timeZone: "America/New_York")' \
        ' |> filter(fn: (r) => r._measurement == "{}")' \
        ' |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")' \
        ' |> filter(fn: (r) => r["ticker"] == "AAPL")' \
        ' |> limit(n:10, offset: 0)'.format(bucket, measurement)

# 方式2：单独使用convertTimezone函数转换
query = 'from(bucket:"{}")' \
        ' |> range(start: 0, stop: now())' \
        ' |> filter(fn: (r) => r._measurement == "{}")' \
        ' |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")' \
        ' |> filter(fn: (r) => r["ticker"] == "AAPL")' \
        ' |> convertTimezone(fn: (t) => time(v: t), location: "America/New_York")' \
        ' |> limit(n:10, offset: 0)'.format(bucket, measurement)

2. 写入前将时间转换为UTC（可选）

如果希望存储的时间与原始时间戳的UTC等价值一致，可在写入前将DataFrame索引转换为UTC时区：

df = yf.Ticker('AAPL').history(period="1d")
# 将带时区的索引转换为UTC
df.index = df.index.tz_convert('UTC')

写入后查询到的UTC时间将与原始时间戳的UTC转换值完全匹配，无需额外转换。

3. 确认写入时区识别正常

你的DataFrame索引已为datetime64[ns, tz]类型，InfluxDB Python客户端会自动识别时区并转换为UTC存储，无需额外配置即可保证时间精度。

内容的提问来源于stack exchange，提问作者grayred8