实现彭博公式15分钟间隔定时执行并存储数据以生成日内折线图的技术求助

看起来你已经搭了个初步的代码架子，但在定时逻辑、数据存储和交易时间窗口控制这几块还有可以调整的地方，我来帮你一步步理顺：

先说说你当前代码里的问题 #

• 你用sched库的enterabs方法时，传入的时间参数逻辑不对，只会让任务执行一次，没法实现每15分钟重复运行的需求
• 只做了数据打印，没有持久化存储，这样后续根本没法生成日内折线图
• 没有限定执行时间在9am到4pm之间，就算定时逻辑对了，程序也会无限制跑下去

改进方案：用schedule库简化定时+增加数据存储+时间窗口控制 #

我给你调整出了一份更贴合需求的代码，每部分都加了注释，你可以直接参考：

import schedule
import time
import datetime
from xbbg import blp
import pandas as pd

# 初始化存储数据的结构，同时用内存DataFrame和本地CSV做双重保险
data_storage = pd.DataFrame(columns=['timestamp', 'yas_yld_spread'])

def action():
    try:
        # 调用彭博API拉取数据
        datapull = blp.bdp(tickers='BBG000BBV9N3 FIGI', flds=['YAS_YLD_SPREAD'])
        datapull_final = round(datapull.yas_yld_spread[0], 2)
        # 获取当前精确时间戳
        current_time = datetime.datetime.now()
        # 把数据存入内存DataFrame
        global data_storage
        data_storage = pd.concat([
            data_storage,
            pd.DataFrame({'timestamp': [current_time], 'yas_yld_spread': [datapull_final]})
        ], ignore_index=True)
        # 同时追加写入本地CSV，防止程序崩溃丢失数据
        data_storage.to_csv('intraday_spread_data.csv', index=False)
        print(f"{current_time.strftime('%H:%M:%S')}: 拉取到利差数据 {datapull_final}")
    except Exception as e:
        # 捕获API调用异常，避免程序直接崩溃
        print(f"{datetime.datetime.now().strftime('%H:%M:%S')}: 彭博API调用失败，错误信息：{str(e)}")

def main():
    # 设定每15分钟执行一次拉取任务
    schedule.every(15).minutes.do(action)
    # 先手动执行一次，不用等第一个15分钟周期
    action()
    
    # 循环检查定时任务，同时控制仅在交易时间内运行
    while True:
        current_hour = datetime.datetime.now().hour
        # 9点到16点（4pm）之间运行，到点自动停止
        if 9 <= current_hour < 16:
            schedule.run_pending()
            time.sleep(1)  # 每秒检查一次，避免占用过多系统资源
        else:
            print("已到当日结束时间，程序停止运行")
            # 最后导出一份完整的当日数据文件
            data_storage.to_csv('intraday_spread_data_final.csv', index=False)
            break

if __name__ == "__main__":
    main()

代码关键部分说明 #

• 定时逻辑：用schedule库替代sched，语法更直观，专门适配这种重复定时任务场景
• 数据存储：同时用内存DataFrame和本地CSV存储，既方便程序运行中调用，又能防止意外崩溃丢失数据
• 时间窗口：在主循环里判断当前小时，严格限定在9am到4pm之间执行，到点自动退出并导出最终数据
• 异常处理：给彭博API调用加了捕获逻辑，就算偶尔API调用失败，程序也不会直接挂掉，能继续执行后续任务

后续生成日内折线图的方法 #

等当日任务跑完后，你可以用这段代码读取存储的CSV文件生成折线图：

import pandas as pd
import matplotlib.pyplot as plt

# 读取当日存储的数据
df = pd.read_csv('intraday_spread_data_final.csv')
# 把时间列转换为可识别的时间格式
df['timestamp'] = pd.to_datetime(df['timestamp'])

# 生成折线图
plt.figure(figsize=(12, 6))
plt.plot(df['timestamp'], df['yas_yld_spread'], marker='o', linestyle='-', color='#1f77b4')
plt.title('日内YAS利差变化折线图')
plt.xlabel('时间')
plt.ylabel('利差（%）')
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()

备注：内容来源于stack exchange，提问作者rocket