目录

简介

一、功能概述

二、环境准备

1. 依赖安装

2. 接口权限说明

三、完整代码实现

四、核心逻辑解析

1. 数据结构化设计

2. 分页请求逻辑

3. 时间筛选机制

4. 容错与稳定性设计

五、使用与扩展说明

1. 参数调整建议

2. 功能扩展方向

3. 常见问题处理

六、参考资源

---

Polymarket 的 Gamma API 提供了丰富的市场数据接口，支持开发者获取活跃市场信息、价格波动、到期时间等核心数据。本文将详细介绍如何通过 API 批量拉取多页市场数据，并筛选出指定时间内到期的市场，代码逻辑清晰、可直接集成到实际项目中。

一、功能概述

本方案通过 Gamma API 实现两大核心功能：

• 多页批量获取 Polymarket 活跃市场数据，支持分页控制与请求频率限制
• 筛选指定时间阈值内即将到期的活跃市场，提取关键信息供后续分析或策略使用

核心优势：采用标准分页参数（limit/offset）确保数据完整性，通过时间范围过滤减少无效请求，数据结构化映射便于后续处理。

二、环境准备

1. 依赖安装

核心依赖为 httpx（HTTP 请求库）和 python-dateutil（时间解析工具），执行以下命令安装：

pip install httpx python-dateutil

• httpx：替代 requests，支持更稳定的异步请求与超时控制；
• python-dateutil：处理 API 返回的 ISO 格式时间字符串，兼容多时区解析。

2. 接口权限说明

Gamma API 的市场列表接口为公开访问，无需身份验证（无需 API 密钥或令牌），直接请求即可获取数据。若需访问账户相关接口（如订单、余额），需参考 Polymarket 官方文档配置授权。

三、完整代码实现

import httpx
import logging
import time
from datetime import datetime, timedelta
from dateutil import parser

# 配置日志：输出请求状态与错误信息，便于排查问题
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

# Gamma API 核心配置（固定参数）
GAMMA_MARKETS_API = "https://gamma-api.polymarket.com/markets"  # 市场数据接口端点
REQUEST_HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36",
    "Accept": "application/json, text/plain, */*",
    "Accept-Language": "en-US,en;q=0.9",
    "Cache-Control": "no-cache",
    "Pragma": "no-cache",
    "Connection": "keep-alive"
}  # 模拟浏览器请求头，避免被接口限流


class SimpleMarket:
    """结构化市场数据对象，提取核心字段便于使用"""
    def __init__(self, **kwargs):
        self.id = kwargs.get("id")  # 市场唯一标识
        self.question = kwargs.get("question")  # 市场预测问题
        self.end_date = kwargs.get("end")  # 到期时间（ISO格式字符串）
        self.active = kwargs.get("active")  # 是否处于活跃状态
        self.spread = kwargs.get("spread")  # 买卖价差
        self.outcome_prices = kwargs.get("outcome_prices")  # 标的价格（YES/NO 选项价格）
        self.volume = kwargs.get("volume")  # 交易量
        self.liquidity = kwargs.get("liquidity")  # 市场流动性


def map_api_to_market(market_data) -> SimpleMarket:
    """将 API 返回的原始字典数据映射为 SimpleMarket 对象"""
    return SimpleMarket(
        id=market_data.get("id"),
        question=market_data.get("question"),
        end=market_data.get("endDate"),  # 接口返回的到期时间字段为 endDate
        active=market_data.get("active"),
        spread=market_data.get("spread"),
        outcome_prices=market_data.get("outcomePrices"),  # 标的价格字段为 outcomePrices
        volume=market_data.get("volume"),
        liquidity=market_data.get("liquidity")
    )


def get_latest_markets_from_gamma():
    """从 Gamma API 多页拉取活跃市场数据，支持分页控制与请求容错"""
    try:
        logger.info(f"启动 Gamma API 数据拉取：{GAMMA_MARKETS_API}")

        # 构建时间筛选条件：仅获取「当前时间 ~ 未来24小时」内到期的市场
        current_utc = datetime.utcnow()
        current_time = current_utc.isoformat() + "Z"  # 转为 UTC 时间格式（API 标准要求）
        expiry_time = (current_utc + timedelta(hours=24)).isoformat() + "Z"

        all_markets = []
        offset = 0  # 分页起始偏移量
        limit = 50  # 单页获取数量（建议值：50-100，避免单页数据过大导致超时）
        max_total = 5000  # 最大拉取总量（防止无限循环）

        while len(all_markets) < max_total:
            # 分页请求参数配置
            params = {
                "active": "true",  # 仅获取活跃市场
                "closed": "false",  # 排除已关闭市场
                "archived": "false",  # 排除已归档市场
                "endDate__gt": current_time,  # 到期时间 > 当前时间（未过期）
                "endDate__lt": expiry_time,  # 到期时间 < 未来24小时（即将到期）
                "limit": limit,
                "offset": offset
            }

            # 发送 HTTP GET 请求
            response = httpx.get(
                url=GAMMA_MARKETS_API,
                headers=REQUEST_HEADERS,
                params=params,
                timeout=10  # 超时时间：10秒（避免请求阻塞）
            )
            response.raise_for_status()  # 触发 HTTP 错误（如 404、500）

            raw_data = response.json()
            current_page_count = len(raw_data)
            logger.debug(f"第 {offset//limit + 1} 页：返回 {current_page_count} 条市场数据")

            # 解析当前页数据并添加到总列表
            page_markets = [map_api_to_market(item) for item in raw_data]
            all_markets.extend(page_markets)

            # 终止条件：当前页数据不足 limit，说明已无更多数据
            if current_page_count < limit:
                break

            # 准备下一页请求
            offset += limit
            time.sleep(1)  # 请求间隔：1秒（避免高频请求触发限流）

        logger.info(f"数据拉取完成：共获取 {len(all_markets)} 个符合条件的活跃市场")
        return all_markets

    except httpx.RequestError as e:
        logger.error(f"API 请求失败（网络/超时问题）：{str(e)}", exc_info=True)
    except httpx.HTTPStatusError as e:
        logger.error(f"API 响应错误（状态码：{e.response.status_code}）：{e.response.text}", exc_info=True)
    except Exception as e:
        logger.error(f"数据解析失败：{str(e)}", exc_info=True)
    return []


def filter_near_expiry_markets(markets, hours_threshold=24):
    """筛选指定时间阈值内即将到期的市场（二次校验，确保数据准确性）"""
    current_time = datetime.now().astimezone()  # 本地时区（适配不同环境）
    expiry_threshold = current_time + timedelta(hours=hours_threshold)
    near_expiry_markets = []

    for market in markets:
        try:
            # 解析 ISO 格式到期时间（兼容多时区）
            market_end_time = parser.isoparse(market.end_date).astimezone()
            # 筛选条件：市场活跃 + 到期时间在阈值范围内
            if market.active and current_time < market_end_time < expiry_threshold:
                near_expiry_markets.append(market)
        except ValueError as e:
            logger.warning(f"市场 {market.id} 时间解析失败：{str(e)}（跳过该条数据）")

    return near_expiry_markets


def test_gamma_api():
    """测试流程：拉取数据 → 筛选到期市场 → 输出结果"""
    # 1. 批量拉取活跃市场数据
    active_markets = get_latest_markets_from_gamma()
    if not active_markets:
        logger.error("未获取到任何活跃市场数据")
        return

    # 2. 筛选24小时内到期的市场
    near_expiry_markets = filter_near_expiry_markets(active_markets, hours_threshold=24)

    # 3. 输出筛选结果
    logger.info(f"\n=== 24小时内到期的活跃市场（共 {len(near_expiry_markets)} 个）===")
    for idx, market in enumerate(near_expiry_markets[:5], 1):  # 仅展示前5个（避免输出过长）
        remaining_time = parser.isoparse(market.end_date).astimezone() - datetime.now().astimezone()
        logger.info(
            f"\n{idx}. 市场ID：{market.id}\n"
            f"   问题：{market.question[:60]}...\n"
            f"   到期时间：{market.end_date}\n"
            f"   剩余时间：{remaining_time}\n"
            f"   流动性：{market.liquidity}\n"
            f"   最新价差：{market.spread}"
        )

    logger.info("\n=== 数据处理流程完成 ===")


if __name__ == "__main__":
    test_gamma_api()

四、核心逻辑解析

1. 数据结构化设计

SimpleMarket 类用于提取 API 响应中的核心字段，避免直接操作原始字典带来的键名错误风险。核心字段说明：

• id：市场唯一标识（用于后续查询市场详情、价格趋势等）；
• outcome_prices：标的价格（字典格式，包含 YES/NO 选项的当前价格）；
• liquidity：市场流动性（数值越高，交易滑点越小）；
• spread：买卖价差（数值越小，市场效率越高）。

2. 分页请求逻辑

采用 offset + limit 标准分页方案（替代非标准的 id__lt 分页），优势如下：

• 兼容性更强：符合 REST API 设计规范，适配多数接口；
• 可控性更高：通过 offset 精确控制起始位置，避免数据遗漏；
• 终止条件清晰：当单页返回数据不足 limit 时，判定为无更多数据。

3. 时间筛选机制

• 接口层面：通过 endDate__gt 和 endDate__lt 参数提前过滤时间范围，减少无效数据传输；
• 本地层面：通过 filter_near_expiry_markets 二次校验，避免因时区差异导致的筛选误差。

4. 容错与稳定性设计

• 超时控制：timeout=10 避免请求长期阻塞；
• 异常捕获：区分网络错误、HTTP 错误、解析错误，便于定位问题；
• 请求间隔：time.sleep(1) 降低高频请求导致的限流风险；
• 总量限制：max_total=5000 防止因接口逻辑变更导致的无限循环。

五、使用与扩展说明

1. 参数调整建议

• limit：单页获取数量，根据网络状况调整（网络稳定可设为 100，不稳定设为 30）；
• max_total：最大拉取总量，按需调整（如仅需近期数据，可设为 500）；
• hours_threshold：到期时间阈值，支持自定义（如筛选 6 小时内到期的市场，设为 6）。

2. 功能扩展方向

• 市场详情查询：通过 market.id 调用 /markets/{id} 接口，获取更详细的价格历史、交易量趋势；
• 价格监控：定时拉取数据，对比 outcome_prices 变化，触发价格波动告警；
• 数据存储：将 SimpleMarket 对象序列化（JSON/CSV），存储到本地文件或数据库；
• 多线程拉取：对于需要大量数据的场景，可通过多线程并行拉取不同分页（需控制并发数，避免限流）。

3. 常见问题处理

• 接口限流：若出现 429 状态码，增加 time.sleep 间隔（如改为 2 秒），或减少 limit 数值；
• 时间解析失败：检查 market.end_date 格式是否为 ISO 标准（如 2024-12-31T23:59:59Z），可添加格式兼容处理；
• 数据为空：确认 active、closed、archived 参数是否正确，或扩大时间筛选范围（如 timedelta(hours=48)）；
• 超时错误：降低 limit 数值，或增加 timeout 时间（如 15 秒）。

六、参考资源

• Polymarket Gamma API 官方文档：https://docs.polymarket.com/
• httpx 官方文档：https://www.python-httpx.org/
• python-dateutil 时间解析文档：https://dateutil.readthedocs.io/

以上代码可直接运行，如需集成到项目中，建议根据实际需求调整参数与错误处理逻辑。若需扩展特定功能（如数据存储、价格监控），可基于现有结构进行模块化开发。