用 Python 留存市场研究依据

#!/usr/bin/env python3
"""Read-only research example. Python 3.11+, standard library, paid read API Key.

POLYSTADIO_API_KEY is read only from the environment and is never printed.
Historical identifiers and a successful prior capture do not promise a live book.
"""
import argparse
import base64
import hashlib
import json
import os
from pathlib import Path
from urllib.error import HTTPError
from urllib.request import Request, urlopen

ORIGIN = 'https://api.polystadio.com'


def call(operation, payload, output):
    request = Request(ORIGIN + '/v1/market-data/polymarket/' + operation,
                      data=json.dumps(payload).encode(), method='POST',
                      headers={'Authorization': 'Bearer ' + os.environ['POLYSTADIO_API_KEY'],
                               'Content-Type': 'application/json'})
    try:
        response = urlopen(request, timeout=25)
    except HTTPError as error:
        response = error
    with response:
        raw = response.read()
        status = response.status
    # Exact service response bytes, before json.loads or reserialization.
    (output / (operation + '.response.json')).write_bytes(raw)
    result = json.loads(raw)
    for receipt in result['evidence']:
        if receipt['body_base64'] is not None:
            body = base64.b64decode(receipt['body_base64'], validate=True)
            assert len(body) == receipt['body_byte_length']
            assert hashlib.sha256(body).hexdigest() == receipt['body_sha256']
    print(json.dumps({'http_status': status, 'request_id': result['request_id'],
                      'error': result['error'], 'upstream': result['upstream_summary']}, ensure_ascii=False))
    if status != 200:
        code = (result.get('error') or {}).get('code')
        if code == 'MARKET_BINDING_MISMATCH':
            raise SystemExit('市场与结果标识不匹配，请重新核对搜索结果，不要跨市场拼接盘口。')
        if status in (401, 403):
            raise SystemExit('请核对Key权限、套餐有效期和服务端环境变量；API Key不能替代网站登录。')
        raise SystemExit('调用未成功，请用请求编号在控制台核对；不要盲目循环重试。')
    return result


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('query', help='事件关键词；搜索有界，不保证穷尽全部市场')
    parser.add_argument('--market-id', help='从搜索结果明确选择市场，不能仅凭名称猜测')
    parser.add_argument('--outcome-id', help='从选中市场的outcomes明确选择结果资产')
    parser.add_argument('--output', type=Path, default=Path('research-evidence'))
    args = parser.parse_args()
    if not os.environ.get('POLYSTADIO_API_KEY'):
        parser.error('请先在本机安全设置环境变量 POLYSTADIO_API_KEY，不要将Key发到聊天或写入源码。')
    args.output.mkdir(parents=True, exist_ok=True)
    search = call('search', {'query': args.query, 'limit': 10}, args.output)
    candidates = search['data']['candidates']
    print(json.dumps(candidates, ensure_ascii=False, indent=2))
    selected = [c for c in candidates if c['market_id'] == args.market_id]
    if len(selected) != 1 or args.outcome_id not in {o['outcome_id'] for o in selected[0]['outcomes']}:
        raise SystemExit('搜索已保存。请核对上述事件、市场和结果后，传入--market-id与--outcome-id重新运行。')
    market = selected[0]
    snapshot = call('snapshot', {'market_id': market['market_id'], 'outcome_id': args.outcome_id,
                                'identifier_schema': market['identifier_schema']}, args.output)
    data = snapshot['data']
    assert data['binding_status'] == 'matched'
    print(json.dumps({'market': data['market']['question'], 'event_ids': data['market']['event_ids'],
                      'selected_outcome_id': data['market']['selected_outcome_id'],
                      'source_book_at': data['orderbook']['source_book_at'],
                      'observed_at': data['collection_completed_at'],
                      'levels': data['orderbook']['returned_count']}, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()