如何用Python将两个字典转换为InfluxDB写入的Measurement字段

问题背景 #

你有两个字典数据：

• 资产金额字典：键为资产名称（如BTC、ETH），值为子字典，子字典键为到期日，值为对应金额
• 到期日类型字典：键为到期日，值为该日期对应的过期类型列表

需要批量转换为符合InfluxDB要求的写入数据结构，单个正确的点结构应为：

{
    'measurement': 'measurement_name',
    'tags': {
        'asset': 'BTC', 
        'maturity_date': '030323', 
        'Expiry': 'weekly'
    },
    'fields': {
        'Amount': 562.7 
    },
    'time': int(time.time()*1000),
}

（注：你提供的示例中存在语法错误，已将-修正为字典键值对的:，同时金额建议保留数值类型而非字符串，更符合InfluxDB字段类型规范）

批量处理实现代码 #

import time

# 定义你的两个原始字典
asset_amounts = {
    'BTC': {'030323': 562.7,
            '170323': 9.900000000000002,
            '250223': 64.60000000000004,
            '290923': 17.6,
            '100323': 53.90000000000002,
            '310323': 27.4,
            '260223': 15.499999999999996,
            '300623': 33.7,
            '280423': 30.0,
            '291223': 28.6,
            '260523': 1.0999999999999999},
    'ETH': {'030323': 562.7,
            '170323': 9.900000000000002,
            '250223': 64.60000000000004,
            '290923': 17.6,
            '100323': 53.90000000000002,
            '310323': 27.4,
            '260223': 15.499999999999996,
            '300623': 33.7,
            '280423': 30.0,
            '291223': 28.6,
            '260523': 1.0999999999999999}
}

expiry_types = {
    '250223': ['daily'],
    '260223': ['bidaily'],
    '030323': ['weekly'],
    '100323': ['biweekly'],
    '170323': ['triweekly'],
    '310323': ['monthly', 'quarterly'],
    '280423': ['bimonthly'],
    '260523': ['trimonthly'],
    '300623': ['biquarterly'],
    '290923': ['triquarterly'],
    '291223': ['yearly']
}

# 定义measurement名称
MEASUREMENT_NAME = 'asset_expiry_amounts'
# 初始化存储所有InfluxDB点的列表
influx_points = []

# 批量生成数据结构
for asset, date_amounts in asset_amounts.items():
    for maturity_date, amount in date_amounts.items():
        # 获取当前到期日对应的所有过期类型
        expiries = expiry_types.get(maturity_date, [])
        # 每个过期类型生成一个独立的InfluxDB点
        for expiry in expiries:
            point = {
                'measurement': MEASUREMENT_NAME,
                'tags': {
                    'asset': asset,
                    'maturity_date': maturity_date,
                    'Expiry': expiry
                },
                'fields': {
                    'Amount': amount
                },
                'time': int(time.time() * 1000)
            }
            influx_points.append(point)

# 可以打印前几个点验证结果
print(influx_points[:3])

关键逻辑说明 #

• 多层循环映射：通过两层循环遍历资产和对应的到期日/金额，再嵌套循环处理单个到期日对应的多个过期类型
• 多标签处理：针对像310323这类有多个过期类型的日期，为每个类型生成独立的数据点，确保InfluxDB的维度标签准确
• 数据类型规范：保留金额的数值类型（而非字符串），符合InfluxDB对数值字段的存储要求，方便后续聚合计算
• 时间戳统一：所有数据点使用当前时间的毫秒级时间戳，也可根据需求替换为业务相关的时间

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