Python API循环获取超量时间索引DataFrame及优化方案问询

针对你提出的脚本不稳定、拉取效率低，以及NaN处理导致样本量不足的问题，以下是具体的优化方向和代码改进建议：

核心问题分析 #

1. Token重复获取：每次调用get_data都重新登录获取Token，不仅浪费资源，还可能触发API频率限制，导致脚本不稳定。
2. API请求逻辑冗余：重复构建请求头、错误提取对象ID（objID = str(r_dict)会把整个JSON转为字符串，导致URL拼接错误）、未利用API分页机制，这些都是不稳定和低效的根源。
3. NaN处理滞后：拉取所有列后再统一删除含NaN的行，造成大量无效API请求和数据处理开销。

一、优化Token管理，避免重复登录 #

将Token缓存起来，仅在过期时重新获取，减少不必要的登录请求：

import time
import pytz
import pandas as pd
import requests
from concurrent.futures import ThreadPoolExecutor

# 全局缓存Token和过期时间
_token_cache = {
    'token': None,
    'expire_time': 0
}

def get_metasys_token():
    current_time = time.time()
    # 检查Token是否有效
    if _token_cache['token'] and current_time < _token_cache['expire_time']:
        return _token_cache['token']
    
    # 重新登录获取Token
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/vnd.metasysapi.v4+json'
    }
    url = 'https://*IP ADDRESS*/api/v4/login'
    body = '{"username": "USER", "password": "PASSWORD"}'
    r = requests.post(url, data=body, headers=headers, verify=False, timeout=5)
    r.raise_for_status()  # 主动抛出请求错误，便于排查问题
    r_dict = r.json()
    
    # 更新缓存，expires是秒数，转换为时间戳
    _token_cache['token'] = r_dict['accessToken']
    _token_cache['expire_time'] = current_time + r_dict['expires']
    return _token_cache['token']

二、改进数据拉取逻辑，提升稳定性与效率 #

1. 修复对象ID提取，复用请求头 #

现有代码中objID = str(r_dict)是错误的，应提取JSON中的id字段；同时复用请求头，减少冗余代码：

def get_object_id(fqr, token):
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {token}'
    }
    url = f'https://*IP ADDRESS*/api/v4/objectIdentifiers?fqr={fqr}'
    r = requests.get(url, headers=headers, verify=False)
    r.raise_for_status()
    return r.json()['id']  # 正确提取对象ID

2. 利用API分页拉取单FQR的完整数据 #

Metasys API支持page参数分页，无需手动按时间切分，直接循环分页获取单FQR的所有数据：

def fetch_fqr_data(fqr, start_utc, end_utc, token, page_size=10000):
    obj_id = get_object_id(fqr, token)
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {token}'
    }
    all_items = []
    page = 1
    
    while True:
        trend_url = (
            f'https://*IP ADDRESS*/api/v4/objects/{obj_id}/trendedAttributes/presentValue/samples'
            f'?startTime={start_utc}&endTime={end_utc}&page={page}&pageSize={page_size}&sort=timestamp'
        )
        r = requests.get(trend_url, headers=headers, verify=False)
        r.raise_for_status()
        data = r.json()
        items = data.get('items', [])
        if not items:
            break
        all_items.extend(items)
        page += 1
    
    # 转换为DataFrame
    if not all_items:
        return pd.DataFrame(columns=['Time', fqr])
    df_raw = pd.json_normalize(all_items)
    df_temp = df_raw[['timestamp', 'value.value']].rename(
        columns={'value.value': fqr, 'timestamp': 'Time'}
    )
    df_temp['Time'] = pd.to_datetime(df_temp['Time'])
    df_temp.set_index('Time', inplace=True)
    return df_temp

3. 并行拉取多FQR数据 #

使用线程池并行拉取多个FQR的数据，大幅减少总等待时间：

def get_data(fqr_list, start, end, dropnan):
    token = get_metasys_token()
    # 转换时间为UTC字符串
    start_utc = start.astimezone(pytz.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    end_utc = end.astimezone(pytz.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    
    # 并行拉取所有FQR数据
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = [
            executor.submit(fetch_fqr_data, fqr, start_utc, end_utc, token)
            for fqr in fqr_list
        ]
        # 收集所有结果
        df_list = [future.result() for future in futures]
    
    # 合并DataFrame
    df_data = pd.concat(df_list, axis=1)
    # 处理时区和列名
    df_data.index = df_data.index.tz_convert('US/Central').tz_localize(None)
    df_data.columns = df_data.columns.str.replace("BCONTROLS1:", "", regex=True)
    
    # 按需删除含NaN的行
    if dropnan:
        df_data.dropna(inplace=True)
    
    return df_data

三、提前过滤无效时间点，减少NaN相关开销 #

如果你的业务允许以某个FQR的时间序列为基准（比如选择采样最密集的FQR），可以先拉取基准FQR的数据，然后其他FQR仅保留基准时间点的数据，提前过滤无效行：

def get_data_with_base_fqr(base_fqr, fqr_list, start, end, dropnan):
    token = get_metasys_token()
    start_utc = start.astimezone(pytz.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    end_utc = end.astimezone(pytz.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    
    # 先拉取基准FQR的数据作为索引
    base_df = fetch_fqr_data(base_fqr, start_utc, end_utc, token)
    if base_df.empty:
        return pd.DataFrame()
    
    # 拉取其他FQR并对齐基准索引
    other_fqrs = [fqr for fqr in fqr_list if fqr != base_fqr]
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = [
            executor.submit(fetch_fqr_data, fqr, start_utc, end_utc, token)
            for fqr in other_fqrs
        ]
        other_dfs = [future.result() for future in futures]
    
    # 合并并对齐基准索引
    df_data = base_df.copy()
    for df in other_dfs:
        df_data = df_data.join(df, how='left')
    
    # 处理时区和列名
    df_data.index = df_data.index.tz_convert('US/Central').tz_localize(None)
    df_data.columns = df_data.columns.str.replace("BCONTROLS1:", "", regex=True)
    
    if dropnan:
        df_data.dropna(inplace=True)
    
    return df_data

四、简化全量数据拉取逻辑 #

由于已经实现了分页拉取单FQR的完整数据，无需再用all_data函数循环按时间切分，直接调用get_data即可获取指定时间范围内的全量数据：

def fetch_full_data(fqr_list, start, end, dropnan=False):
    print(f"Start: {start}")
    print(f"End: {end}")
    print(f"Drop NaN: {dropnan}")
    
    df = get_data(fqr_list, start, end, dropnan)
    # 去重（如果有重复时间点）
    df = df.drop_duplicates()
    
    return {
        'data': df,
        'idx': list(df.index),
        'hdr': list(df.columns)
    }

关键优化点总结 #

• Token缓存：减少登录请求，避免API限制，提升稳定性。
• 分页拉取：利用API原生分页，替代手动时间切分，逻辑更可靠。
• 并行请求：多线程拉取多个FQR，大幅提升拉取效率。
• 提前对齐索引：以基准FQR的时间为核心，减少无效数据的拉取和处理，降低NaN带来的开销。

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