Pharos
Global

HttpQuery - HTTP 请求

发送 HTTP 请求并获取响应。

语法

python
# GET 请求
response = HttpQuery(url)

# 带选项的请求
response = HttpQuery(url, options)

参数

参数名类型必选默认值说明
urlstring是无请求的 URL 地址
optionsdict否{}请求选项

options 选项

字段类型默认值说明
methodstring'GET'请求方法：GET, POST, PUT, DELETE
headersdict{}请求头字典
bodystringNone请求体数据
timeoutnumber10000超时时间（毫秒）
debugboolFalse是否返回详细信息

返回值

  • 普通模式 (debug=False): 返回响应体字符串（str）
  • 调试模式 (debug=True): 返回包含完整响应信息的字典
    python
    {
    'status_code': 200,
    'headers': {...},
    'body': '...'
}
  • 失败: 返回 None

示例

示例1: 简单 GET 请求

python
# 发送 GET 请求
response = HttpQuery("https://api.example.com/ticker")

if response:
    import json
    data = json.loads(response)
    Log(f"价格: {data['price']}")
else:
    Log("请求失败")

示例2: 带参数的 GET 请求

python
# 构建带查询参数的 URL
symbol = "BTCUSDT"
url = f"https://api.binance.com/api/v3/ticker/price?symbol={symbol}"

response = HttpQuery(url)
if response:
    import json
    data = json.loads(response)
    Log(f"{symbol} 价格: {data['price']}")

示例3: POST 请求

python
import json

# 准备数据
data = {
    "symbol": "BTCUSDT",
    "side": "BUY",
    "quantity": "0.001"
}

# 发送 POST 请求
response = HttpQuery("https://api.example.com/order", {
    "method": "POST",
    "headers": {
        "Content-Type": "application/json",
        "X-API-KEY": "your_api_key"
    },
    "body": json.dumps(data),
    "timeout": 5000
})

if response:
    result = json.loads(response)
    Log(f"订单ID: {result['orderId']}")

示例4: 自定义请求头

python
# 添加自定义请求头
headers = {
    "User-Agent": "MyTradingBot/1.0",
    "Authorization": "Bearer your_token",
    "Accept": "application/json"
}

response = HttpQuery("https://api.example.com/data", {
    "headers": headers,
    "timeout": 10000
})

if response:
    Log(f"响应: {response}")

示例5: 调试模式

python
# 启用调试模式获取完整响应信息
response = HttpQuery("https://api.example.com/test", {
    "debug": True
})

if response:
    Log(f"状态码: {response['status_code']}")
    Log(f"响应头: {response['headers']}")
    Log(f"响应体: {response['body']}")
else:
    Log("请求失败")

示例6: 错误处理

python
def safe_http_request(url, max_retries=3):
    """带重试的 HTTP 请求"""
    for i in range(max_retries):
        try:
            response = HttpQuery(url, {"timeout": 5000})
            if response:
                return json.loads(response)
            else:
                Log(f"第 {i+1} 次请求失败，重试...")
                Sleep(1000)
        except Exception as e:
            Log(f"请求异常: {e}")
            Sleep(1000)
    
    Log("达到最大重试次数")
    return None

# 使用
data = safe_http_request("https://api.example.com/data")
if data:
    Log(f"获取数据成功: {data}")

示例7: 查询多个交易所价格

python
def get_multi_exchange_prices(symbol):
    """查询多个交易所的价格"""
    exchanges = {
        "binance": f"https://api.binance.com/api/v3/ticker/price?symbol={symbol}",
        "okx": f"https://www.okx.com/api/v5/market/ticker?instId={symbol}",
    }
    
    prices = {}
    for exchange_name, url in exchanges.items():
        response = HttpQuery(url, {"timeout": 3000})
        if response:
            try:
                data = json.loads(response)
                if exchange_name == "binance":
                    prices[exchange_name] = float(data['price'])
                elif exchange_name == "okx":
                    prices[exchange_name] = float(data['data'][0]['last'])
            except Exception as e:
                Log(f"{exchange_name} 价格解析失败: {e}")
    
    return prices

# 使用
prices = get_multi_exchange_prices("BTCUSDT")
for exchange, price in prices.items():
    Log(f"{exchange}: {price}")

示例8: Webhook 通知

python
def send_webhook_notification(message, webhook_url):
    """发送 Webhook 通知"""
    import json
    
    payload = {
        "text": message,
        "timestamp": _D()
    }
    
    response = HttpQuery(webhook_url, {
        "method": "POST",
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps(payload),
        "timeout": 5000
    })
    
    if response:
        Log("✅ 通知发送成功")
        return True
    else:
        Log("❌ 通知发送失败")
        return False

# 使用
webhook_url = "https://hooks.example.com/services/YOUR_WEBHOOK"
send_webhook_notification("策略已启动", webhook_url)

示例9: 获取汇率

python
def get_exchange_rate(from_currency, to_currency):
    """获取汇率"""
    url = f"https://api.exchangerate-api.com/v4/latest/{from_currency}"
    
    response = HttpQuery(url)
    if response:
        data = json.loads(response)
        rate = data['rates'].get(to_currency)
        if rate:
            Log(f"{from_currency} to {to_currency}: {rate}")
            return rate
    
    Log("获取汇率失败")
    return None

# 使用
rate = get_exchange_rate("USD", "CNY")
if rate:
    usd_amount = 100
    cny_amount = usd_amount * rate
    Log(f"{usd_amount} USD = {cny_amount} CNY")

注意事项

  1. 超时设置: 默认超时 10 秒，建议根据网络状况调整

  2. 错误处理: 请求失败返回 None，应始终检查返回值

  3. HTTPS: 支持 HTTP 和 HTTPS 协议

  4. 请求频率: 注意 API 限流，避免请求过于频繁

  5. 响应解析:

    • JSON 响应需要使用 json.loads() 解析
    • 文本响应直接使用字符串

  6. 请求体: POST/PUT 请求的 body 应为字符串，JSON 需先序列化

  7. 超时单位: timeout 参数单位为毫秒

  8. 同步调用: 此函数为同步调用，会阻塞直到请求完成或超时

相关 API

  • Dial - TCP/WebSocket 连接
  • Sleep - 休眠指定毫秒数

GetOS - 获取操作系统信息

Previous Page

IsVirtual - 判断是否为回测环境

Next Page

On this page

HttpQuery - HTTP 请求语法参数options 选项返回值示例示例1: 简单 GET 请求示例2: 带参数的 GET 请求示例3: POST 请求示例4: 自定义请求头示例5: 调试模式示例6: 错误处理示例7: 查询多个交易所价格示例8: Webhook 通知示例9: 获取汇率注意事项相关 API