1 Star 0 Fork 0

jisys / quantsys-sdk

加入 Gitee
与超过 800 万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
克隆/下载
取消
提示: 由于 Git 不支持空文件夾,创建文件夹后会生成空的 .keep 文件
Loading...
README.md

yufuquant SDK

yufuquant SDK 封装了用于和 yufuquant 后端进行交互的常用方法。

yufuquant SDK 目前只支持 Python,由于大多数 API 都是基于 asyncio 的异步 API,因此要求 Python 的最低版本为 Python 3.7,推荐使用 Python 3.8。

安装

$ pip install quantsyssdk

REST API 客户端

REST API 客户端用于和 yufuquant 后端的 RESTful API 进行交互。

from quantsyssdk.clients import RESTAPIClient

base_url = "https://yufuquant.cc/api/v1"  # 系统后端接口地址
auth_token = "xxxxx"  # 认证令牌
robot_id = 1  # 机器人 id
rest_api_client = RESTAPIClient(base_url=base_url, auth_token=auth_token)

# 获取机器人配置
await rest_api_client.get_robot_config(robot_id)

# 更新机器人的资产信息
data = {
    "total_balance": 5.8
}
await rest_api_client.patch_robot_asset_record(robot_id, data)

# 发送 ping
await rest_api_client.post_robot_ping(robot_id)

以下是一个完整的示例:

import logging
from pprint import pprint
from quantsyssdk.clients import RESTAPIClient

# 开启日志
logger = logging.getLogger("quantsyssdk")
logger.addHandler(logging.StreamHandler())
logger.setLevel(logging.DEBUG)


async def main():
    http_client = RESTAPIClient(
        base_url="https://yufuquant.cc/api/v1",
        auth_token="8g2e470579ba14ea69000859eba6c421b69ff95d",
    )

    result = await http_client.get_robot_config(robot_id=1)
    pprint(result)

    result = await http_client.post_robot_ping(robot_id=1)
    pprint(result)

    result = await http_client.patch_robot_asset_record(
        robot_id=1, data={"total_balance": 10000}
    )
    pprint(result)


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

Websocket API 客户端

Websocket API 客户端用于和 yufuquant 后端的 Websocket API 进行交互。

from quantsyssdk.clients import WebsocketAPIClient

uri = "wss://yufuquant.cc/ws/v1/streams/"  # 系统后端接口地址
auth_token = "xxxxx"  # 认证令牌
topics = ["robot#1.ping", "robot#1.log"]  # 订阅的话题
ws_api_client = WebsocketAPIClient(uri=uri)

# 认证
await ws_api_client.auth(auth_token)

# 订阅话题
await ws_api_client.sub(topics)

# 取消话题订阅
await ws_api_client.unsub(topics)

# 发送机器人 ping
await ws_api_client.robot_ping()

# 发送机器人日志
await ws_api_client.robot_log()

以下是一个完整的示例:

import logging
from quantsyssdk.clients import WebsocketAPIClient

# 开启日志
logger = logging.getLogger("quantsyssdk")
logger.addHandler(logging.StreamHandler())
logger.setLevel(logging.DEBUG)


async def main():
    ws_api_client = WebsocketAPIClient(uri="wss://yufuquant.cc/ws/v1/streams/")

    await ws_api_client.auth("8d2e470575ba04ea69000859eba6c421a69ff95c")
    await ws_api_client.sub(topics=["robot#1.log"])
    while True:
        await ws_api_client.robot_ping()
        await ws_api_client.robot_log("Test robot log...", level="INFO")
        await asyncio.sleep(1)


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

简介

用于连接量化系统的api集合 展开 收起
Python
MIT
取消

发行版

暂无发行版

贡献者

全部

近期动态

加载更多
不能加载更多了
Python
1
https://gitee.com/jisys/quantsys-sdk.git
git@gitee.com:jisys/quantsys-sdk.git
jisys
quantsys-sdk
quantsys-sdk
master

搜索帮助