如何使用Chainlink API提取Avalanche价格喂价2年逐日历史数据用于机器学习

前置准备 #

• 替换示例中的测试网RPC为Avalanche主网RPC，可使用公开主网入口https://api.avax.network/ext/bc/C/rpc，也可使用自己部署的节点RPC提升查询稳定性
• 确认要拉取的交易对对应的Avalanche主网喂价合约地址，例如AVAX/USD的喂价合约地址为0x0A77230d17318075983913bC2145DB16C7366156
• 所有Chainlink喂价都兼容AggregatorV3Interface标准ABI，直接使用通用ABI调用即可，无需单独获取对应合约的ABI文件

核心逻辑说明 #

Chainlink喂价每一次价格更新都会生成唯一的roundId，调用getRoundData返回的结果结构为(roundId, answer, startedAt, updatedAt, answeredInRound)：

• answer为加密货币价格，需要除以合约返回的小数位数（通常为8位，即除以1e8）得到实际法币计价的价格
• updatedAt为该轮价格更新的时间戳，可用来匹配对应的日期
• 逐日价格取当日最后一次更新的喂价数值作为当日收盘价即可，无需存储所有轮次的价格数据

完整实现代码 #

首先安装依赖：

pip install web3 pandas

实现代码示例：

from web3 import Web3
from collections import defaultdict
import datetime
import pandas as pd

# 初始化Avalanche主网连接
w3 = Web3(Web3.HTTPProvider('https://api.avax.network/ext/bc/C/rpc'))

# AggregatorV3Interface 通用ABI
ABI = [
    {"inputs":[],"name":"decimals","outputs":[{"internalType":"uint8","name":"","type":"uint8"}],"stateMutability":"view","type":"function"},
    {"inputs":[{"internalType":"uint80","name":"roundId","type":"uint80"}],"name":"getRoundData","outputs":[{"internalType":"uint80","name":"roundId","type":"uint80"},{"internalType":"int256","name":"answer","type":"int256"},{"internalType":"uint256","name":"startedAt","type":"uint256"},{"internalType":"uint256","name":"updatedAt","type":"uint256"},{"internalType":"uint80","name":"answeredInRound","type":"uint80"}],"stateMutability":"view","type":"function"},
    {"inputs":[],"name":"latestRoundData","outputs":[{"internalType":"uint80","name":"roundId","type":"uint80"},{"internalType":"int256","name":"answer","type":"int256"},{"internalType":"uint256","name":"startedAt","type":"uint256"},{"internalType":"uint256","name":"updatedAt","type":"uint256"},{"internalType":"uint80","name":"answeredInRound","type":"uint80"}],"stateMutability":"view","type":"function"}
]

# 替换为目标交易对的喂价合约地址，示例为AVAX/USD
FEED_CONTRACT_ADDR = "0x0A77230d17318075983913bC2145DB16C7366156"
price_feed_contract = w3.eth.contract(address=FEED_CONTRACT_ADDR, abi=ABI)

# 获取价格小数位数，用于转换实际价格
decimals = price_feed_contract.functions.decimals().call()
price_unit = 10 ** decimals

# 获取最新轮次ID，从最新数据倒序遍历
latest_round_data = price_feed_contract.functions.latestRoundData().call()
current_round_id = latest_round_data[0]

# 计算2年前的时间戳，作为遍历终止条件
two_years_ago_ts = (datetime.datetime.now() - datetime.timedelta(days=730)).timestamp()

daily_price_map = defaultdict(dict)

while True:
    try:
        round_data = price_feed_contract.functions.getRoundData(current_round_id).call()
        _, answer, _, updated_at, _ = round_data
        # 数据早于2年就停止遍历
        if updated_at < two_years_ago_ts:
            break
        # 转换为日期字符串作为去重key
        date_str = datetime.datetime.fromtimestamp(updated_at).strftime("%Y-%m-%d")
        # 同一个日期仅保留最后更新的价格作为当日收盘价
        if date_str not in daily_price_map or updated_at > daily_price_map[date_str]["update_ts"]:
            daily_price_map[date_str] = {
                "price": answer / price_unit,
                "update_ts": updated_at
            }
        # 往前遍历上一个轮次
        current_round_id -= 1
    except:
        # 遇到无效轮次直接跳过
        current_round_id -= 1
        continue

# 导出为CSV文件用于机器学习训练
result_list = [{"date": date, "price": item["price"]} for date, item in daily_price_map.items()]
df = pd.DataFrame(result_list).sort_values("date")
df.to_csv("avalanche_daily_price.csv", index=False)
print(f"成功获取{len(df)}天的历史价格数据")

注意事项 #

• 若查询过程中遇到RPC节点限流，可在每次调用getRoundData后添加0.1秒延迟，或替换为私有Avalanche节点RPC提升查询速度
• 2年的历史数据对应约3~5万条轮次记录，普通公开RPC即可支撑查询，无需额外优化
• 若需要更高精度的小时级/分钟级数据，调整去重逻辑的时间粒度即可

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