Python 示例

ccxt 太重、只想直调 REST 时走原生 /v1/*。

原生 REST

import os, requests

API = 'https://api.panshi.io'
KEY = os.environ['PANSHI_API_KEY']
H   = {'Authorization': f'Bearer {KEY}'}

# 列出我的组合
r = requests.get(f'{API}/v1/portfolios/my', headers=H, timeout=10)
r.raise_for_status()
for pf in r.json()['portfolios']:
    print(pf['short_code'], pf['name'], pf['current_equity'])

下单（panshi 原生信号）

order = {
    'portfolio_id': '<pf_id>',
    'symbol': 'BTCUSDT',
    'action': 'BUY',        # BUY / SELL / CLOSE
    'amount_usd': 200,
}
r = requests.post(
    f'{API}/v1/order',
    headers={**H, 'Idempotency-Key': 'unique-per-intent-uuid'},
    json=order, timeout=10,
)
print(r.json())

带 Idempotency-Key（任意唯一字符串，TTL 24h）可避免重复下单；同 key 命中缓存时直接返回上次响应，不计入限流窗口。

WebSocket（自己的持仓 / 订单推送）

import asyncio, websockets, json, os
# 私频道走浏览器 session cookie；在 Python 脚本里，先用 api_key 换一个 bearer-aware 连接：
# （如果没有 cookie，可用组合级 api_key 访问 portfolio 流的公开子集，或走 REST 轮询）

async def main():
    url = 'wss://api.panshi.io/ws/portfolio/P0000007'
    async with websockets.connect(url, ping_interval=30) as ws:
        while True:
            msg = await ws.recv()
            print(json.loads(msg))

asyncio.run(main())

错误处理

4xx/5xx 响应统一：

{"error":"insufficient_balance","msg":"wallet short by 12.3 USDT"}

薄包装：

def post(path, **kw):
    r = requests.post(f'{API}{path}', headers=H, timeout=10, **kw)
    if not r.ok:
        body = r.json()
        raise RuntimeError(f"{r.status_code} {body.get('error')} {body.get('msg')}")
    return r.json()

短码查询

# 用 P0000007 这种短码定位组合
r = requests.get(f'{API}/v1/portfolios/by_code/P0000007', headers=H)

下一步

• Binance-compatible API — 想用 ccxt 接
• WebSocket
• 速率限制