InfluxDB的安装与Python调用

pip install influxdb-client

from influxdb_client import InfluxDBClient, Point, WritePrecision
from datetime import datetime
from influxdb_client.client.write_api import SYNCHRONOUS

# 配置 InfluxDB 客户端
url = "http://localhost:8086"  
token = "your_token"  # 替换为你的 API Token
org = "your_org"      # 替换为你的组织名
bucket = "your_bucket" # 替换为你的桶名

# 初始化 InfluxDB 客户端
client = InfluxDBClient(url=url, token=token, org=org)
write_api = client.write_api(write_options=SYNCHRONOUS)
query_api = client.query_api()


# 写入期权数据示例
def write_option_data():
    option_contract = "AAPL210416C00125000" # 期权合约代码
    exchange = "NASDAQ" # 交易所
    option_type = "CALL" # 期权类型
    strike_price = 125.00  # 执行价
    market_price = 5.50   # 市场价格

    point = Point("options") \  # 创建一个名为 options 的测量（measurement）
        .tag("contract", option_contract) \ # 添加期权合约代码作为标签（tag）
        .tag("exchange", exchange) \ # 添加交易所作为标签（tag）。
        .tag("type", option_type) \ # 添加期权类型（CALL/PUT）作为标签（tag）
        .field("strike_price", strike_price) \ # 添加执行价作为字段（field）
        .field("market_price", market_price) \ # 添加市场价格作为字段（field）
        .time(datetime.utcnow(), WritePrecision.NS) # 设置数据点的时间戳为当前时间，并指定时间精度为纳秒（NS）

    write_api.write(bucket=bucket, org=org, record=point) # 写入数据
    
# 查询期权数据示例
def query_option_data():
    # from(bucket:"{bucket}"): 从指定的 bucket 中查询数据。
    # |> range(start: -1h): 查询最近一小时内的数据。
    # |> filter(fn: (r) => r._measurement == "options"): 过滤数据，只保留测量名称为 options 的数据。
    query = f'from(bucket:"{bucket}") |> range(start: -1h) |> filter(fn: (r) => r._measurement == "options")'
    tables = query_api.query(query=query, org=org)

    # 使用字典保存字段值
    data = {}
    print(tables)
    for table in tables:
        for record in table.records:
            print(record)
            field_name = record["_field"]
            field_value = record["_value"]
            data[field_name] = field_value
            if "contract" not in data:
                data["contract"] = record["contract"]
            if "type" not in data:
                data["type"] = record["type"]
            if "exchange" not in data:
                data["exchange"] = record["exchange"]

    # 打印结果
    print(
        f"Contract: {data.get('contract')}, Type: {data.get('type')}, Exchange: {data.get('exchange')}, Strike Price: {data.get('strike_price')}, Market Price: {data.get('market_price')}"
    )

# 主函数
if __name__ == "__main__":
    # 写入示例数据
    write_option_data()

    # 查询示例数据
    query_option_data()