pip install easyquotes

pip install "easyquotes[stream]"        # WebSocket 同步推送
pip install "easyquotes[stream-async]"  # WebSocket 异步推送

export EASYQUOTE_API_KEY="eq_your_key_here"

export EASYQUOTE_BASE_URL="https://quotes.example.com"

from easyquote import EasyQuoteClient

client = EasyQuoteClient()  # 自动读取 EASYQUOTE_API_KEY
quote = client.get_quote("600519.SH")
print(f"{quote.name}: {quote.price} ({quote.change_pct:+.2f}%)")

client = EasyQuoteClient(api_key="eq_your_key_here")

opportunities = client.get_strategy_opportunities(limit=20)
signals = client.get_strategy_signals(state="triggered")

report = client.run_strategy_backtest(
    "2026-07-20",
    "2026-07-21",
    initial_cash=1_000_000,
)

# 模拟账户需要主动创建；不会连接券商或开启实盘交易
paper = client.create_paper_account("研究账户", initial_cash=1_000_000)
detail = client.get_paper_account(paper["id"])

{
  "mcpServers": {
    "easyquote": {
      "url": "https://122.51.7.196/mcp",
      "headers": {
        "x-api-key": "eq_your_key_here"
      }
    }
  }
}

{
  "mcpServers": {
    "easyquote": {
      "command": "npx",
      "args": [
        "-y",
        "mcp-remote",
        "https://122.51.7.196/sse",
        "--header",
        "x-api-key: eq_your_key_here"
      ]
    }
  }
}

curl -H "x-api-key: eq_your_key_here" \
  "https://122.51.7.196/api/v1/etf/stock-holders-db?code=600519"

curl -G -H "x-api-key: eq_your_key_here" \
  --data-urlencode "change=新进" \
  --data-urlencode "hold_type=基金" \
  "https://122.51.7.196/api/v1/fund/new-positions"

quotes = client.get_quotes(["600519.SH", "000001.SZ", "TSLA.US", "00700.HK"])
for q in quotes:
    print(f"{q.symbol}  {q.price:>10.2f}  {q.change_pct:+.2f}%")

klines = client.get_klines("600519.SH", period="daily", count=120)
for k in klines[-5:]:
    print(f"{k.datetime}  O:{k.open}  H:{k.high}  L:{k.low}  C:{k.close}  V:{k.volume}")

bars = client.get_minute_history("600519.SH", date="20260617")
for b in bars[:10]:
    print(b.time, b.price, b.volume)

ff = client.get_fund_flow("600519.SH")
print(f"主力净流入: {ff.net_main / 1e8:.2f} 亿")

ranking = client.get_board_ranking(board_type="industry", top_n=10, sort_by="change_pct")
for b in ranking:
    print(b["name"], b["change_pct"])

with client.stream() as ws:
    ws.subscribe(["600519.SH", "000001.SZ", "TSLA.US"])
    for event in ws:
        if event["op"] == "quotes":
            for q in event["data"]:
                print(q["symbol"], q["last_price"])

import asyncio

async def main():
    async with client.stream_async() as ws:
        await ws.subscribe(["600519.SH", "TSLA.US"])
        async for event in ws:
            print(event)

asyncio.run(main())

ticks = client.get_transactions("600519.SH", start=0, count=50)
for t in ticks:
    print(t["time"], t["price"], t["volume"], t["direction"])

# 获取纳斯达克个股 K 线
klines = client.get_global_klines("US", "TSLA", period="daily", count=60)

# 获取港股行情
quote = client.get_global_quote("HK", "00700")

from easyquote import EasyQuoteClient
from easyquote import AuthError, RateLimitError, APIError, NotFoundError

client = EasyQuoteClient(api_key="eq_your_key")

try:
    quote = client.get_quote("600519.SH")
except AuthError:
    print("API Key 无效或未设置")
except RateLimitError:
    print("请求频率超限，请稍后重试")
except NotFoundError:
    print("标的不存在")
except APIError as e:
    print(f"API 错误 {e.code}: {e.message}")

export EASYQUOTE_BASE_URL="https://quotes.example.com"

client = EasyQuoteClient(api_key="eq_your_key")

client = EasyQuoteClient(
    api_key="eq_your_key",
    base_url="https://quotes.example.com",
)